feat: new TS migration adding renku:slug prop to all Project entities

knowledge-graph/src/test/scala/io/renku/knowledgegraph/projects/datasets/ProjectDatasetSpec.scala

@@ -1,3 +1,21 @@
+/*
+ * Copyright 2023 Swiss Data Science Center (SDSC)
+ * A partnership between École Polytechnique Fédérale de Lausanne (EPFL) and
+ * Eidgenössische Technische Hochschule Zürich (ETHZ).
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package io.renku.knowledgegraph.projects.datasets
 
 import Generators._

triples-generator/src/main/scala/io/renku/triplesgenerator/events/consumers/tsmigrationrequest/migrations/Migrations.scala

@@ -47,6 +47,7 @@ private[tsmigrationrequest] object Migrations {
     provisionProjectsGraph         <- projectsgraph.ProvisionProjectsGraph[F]
     addProjectDateModified         <- datemodified.AddProjectDateModified[F]
     fixMultipleProjectVersions     <- FixMultipleProjectVersions[F]
+    addProjectSlug                 <- projectslug.AddProjectSlug[F]
     migrations <- validateNames(
                     datasetsCreator,
                     datasetsRemover,
@@ -61,7 +62,8 @@ private[tsmigrationrequest] object Migrations {
                     personViewedEntityDeduplicator,
                     provisionProjectsGraph,
                     addProjectDateModified,
-                    fixMultipleProjectVersions
+                    fixMultipleProjectVersions,
+                    addProjectSlug
                   )
   } yield migrations
 

triples-generator/src/main/scala/io/renku/triplesgenerator/events/consumers/tsmigrationrequest/migrations/projectslug/AddProjectSlug.scala

@@ -0,0 +1,112 @@
+/*
+ * Copyright 2023 Swiss Data Science Center (SDSC)
+ * A partnership between École Polytechnique Fédérale de Lausanne (EPFL) and
+ * Eidgenössische Technische Hochschule Zürich (ETHZ).
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.renku.triplesgenerator.events.consumers.tsmigrationrequest.migrations.projectslug
+
+import cats.data.EitherT
+import cats.effect.Async
+import cats.syntax.all._
+import io.renku.metrics.MetricsRegistry
+import io.renku.triplesgenerator.events.consumers.ProcessingRecoverableError
+import io.renku.triplesgenerator.events.consumers.tsmigrationrequest.migrations.tooling.{MigrationExecutionRegister, RecoverableErrorsRecovery}
+import io.renku.triplesgenerator.events.consumers.tsmigrationrequest.{ConditionedMigration, Migration, categoryName}
+import io.renku.triplesstore.{ProjectsConnectionConfig, SparqlQueryTimeRecorder}
+import org.typelevel.log4cats.Logger
+
+private class AddProjectSlug[F[_]: Async: Logger](
+    migrationNeedChecker: MigrationNeedChecker[F],
+    backlogCreator:       BacklogCreator[F],
+    projectsFinder:       ProjectsPageFinder[F],
+    progressFinder:       ProgressFinder[F],
+    projectFetcher:       ProjectFetcher[F],
+    slugPersister:        SlugPersister[F],
+    projectDonePersister: ProjectDonePersister[F],
+    executionRegister:    MigrationExecutionRegister[F],
+    recoveryStrategy:     RecoverableErrorsRecovery = RecoverableErrorsRecovery
+) extends ConditionedMigration[F] {
+
+  import executionRegister._
+  import fs2._
+  import progressFinder._
+  import projectDonePersister._
+  import projectFetcher.fetchProject
+  import projectsFinder._
+  import recoveryStrategy._
+  import slugPersister.persistSlug
+
+  override val name: Migration.Name = AddProjectSlug.name
+
+  protected[projectslug] override def required
+      : EitherT[F, ProcessingRecoverableError, ConditionedMigration.MigrationRequired] = EitherT {
+    migrationNeedChecker.checkMigrationNeeded
+      .map(_.asRight[ProcessingRecoverableError])
+      .recoverWith(maybeRecoverableError[F, ConditionedMigration.MigrationRequired])
+  }
+
+  protected[projectslug] override def migrate(): EitherT[F, ProcessingRecoverableError, Unit] = EitherT {
+    backlogCreator.createBacklog() >>
+      Logger[F].info(show"$categoryName: $name backlog created") >>
+      Stream
+        .iterate(1)(_ + 1)
+        .evalMap(_ => nextProjectsPage())
+        .takeThrough(_.nonEmpty)
+        .flatMap(Stream.emits(_))
+        .evalMap(slug => findProgressInfo.map(slug -> _))
+        .evalTap { case (slug, info) => logInfo(show"provisioning '$slug'", info) }
+        .evalMap { case (slug, info) => fetchProject(slug).map(p => (slug, p, info)) }
+        .evalTap { case (_, maybeProj, _) => maybeProj.map(persistSlug).getOrElse(().pure[F]) }
+        .evalTap { case (slug, _, _) => noteDone(slug) }
+        .evalTap { case (slug, _, info) => logInfo(show"'$slug' provisioned", info) }
+        .compile
+        .drain
+        .map(_.asRight[ProcessingRecoverableError])
+        .recoverWith(maybeRecoverableError[F, Unit])
+  }
+
+  private def logInfo(message: String, progressInfo: String): F[Unit] =
+    Logger[F].info(show"${AddProjectSlug.name} - $progressInfo - $message")
+
+  protected[projectslug] override def postMigration(): EitherT[F, ProcessingRecoverableError, Unit] = EitherT {
+    registerExecution(name)
+      .map(_.asRight[ProcessingRecoverableError])
+      .recoverWith(maybeRecoverableError[F, Unit])
+  }
+}
+
+private[migrations] object AddProjectSlug {
+  val name: Migration.Name = Migration.Name("Add Project slug")
+
+  def apply[F[_]: Async: Logger: MetricsRegistry: SparqlQueryTimeRecorder]: F[Migration[F]] = for {
+    checkMigrationNeeded <- MigrationNeedChecker[F]
+    backlogCreator       <- BacklogCreator[F]
+    projectsFinder       <- ProjectsPageFinder[F]
+    progressFinder       <- ProgressFinder[F]
+    projectFetcher       <- ProjectFetcher[F]
+    datePersister        <- ProjectsConnectionConfig[F]().map(SlugPersister[F](_))
+    projectDonePersister <- ProjectDonePersister[F]
+    executionRegister    <- MigrationExecutionRegister[F]
+  } yield new AddProjectSlug(checkMigrationNeeded,
+                             backlogCreator,
+                             projectsFinder,
+                             progressFinder,
+                             projectFetcher,
+                             datePersister,
+                             projectDonePersister,
+                             executionRegister
+  )
+}

triples-generator/src/main/scala/io/renku/triplesgenerator/events/consumers/tsmigrationrequest/migrations/projectslug/BacklogCreator.scala

@@ -0,0 +1,125 @@
+/*
+ * Copyright 2023 Swiss Data Science Center (SDSC)
+ * A partnership between École Polytechnique Fédérale de Lausanne (EPFL) and
+ * Eidgenössische Technische Hochschule Zürich (ETHZ).
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.renku.triplesgenerator.events.consumers.tsmigrationrequest.migrations
+package projectslug
+
+import cats.effect.{Async, Ref}
+import cats.syntax.all._
+import eu.timepit.refined.auto._
+import io.circe.Decoder
+import io.renku.graph.config.RenkuUrlLoader
+import io.renku.graph.model.Schemas._
+import io.renku.graph.model.{GraphClass, RenkuUrl, projects}
+import io.renku.jsonld.syntax._
+import io.renku.triplesstore.SparqlQuery.Prefixes
+import io.renku.triplesstore._
+import io.renku.triplesstore.client.model.Triple
+import io.renku.triplesstore.client.syntax._
+import org.typelevel.log4cats.Logger
+import tooling.RecordsFinder
+
+private trait BacklogCreator[F[_]] {
+  def createBacklog(): F[Unit]
+}
+
+private object BacklogCreator {
+  def apply[F[_]: Async: Logger: SparqlQueryTimeRecorder]: F[BacklogCreator[F]] = for {
+    implicit0(ru: RenkuUrl) <- RenkuUrlLoader[F]()
+    recordsFinder           <- ProjectsConnectionConfig[F]().map(RecordsFinder[F](_))
+    migrationsDSClient      <- MigrationsConnectionConfig[F]().map(TSClient[F](_))
+  } yield new BacklogCreatorImpl[F](recordsFinder, migrationsDSClient)
+
+  def asToBeMigratedInserts(implicit ru: RenkuUrl): List[projects.Slug] => Option[SparqlQuery] =
+    toTriples andThen toInsertQuery
+
+  private def toTriples(implicit ru: RenkuUrl): List[projects.Slug] => List[Triple] =
+    _.map(slug => Triple(AddProjectSlug.name.asEntityId, renku / "toBeMigrated", slug.asObject))
+
+  private lazy val toInsertQuery: List[Triple] => Option[SparqlQuery] = {
+    case Nil => None
+    case triples =>
+      SparqlQuery
+        .ofUnsafe(
+          show"${AddProjectSlug.name} - store to backlog",
+          sparql"INSERT DATA {\n${triples.map(_.asSparql).combineAll}\n}"
+        )
+        .some
+  }
+}
+
+private class BacklogCreatorImpl[F[_]: Async](recordsFinder: RecordsFinder[F], migrationsDSClient: TSClient[F])(implicit
+    ru: RenkuUrl
+) extends BacklogCreator[F] {
+
+  import BacklogCreator._
+  import io.renku.tinytypes.json.TinyTypeDecoders._
+  import io.renku.triplesstore.ResultsDecoder._
+  import recordsFinder._
+
+  private val pageSize: Int = 50
+
+  override def createBacklog(): F[Unit] = addPageToBacklog(Ref.unsafe(1))
+
+  private def addPageToBacklog(currentPage: Ref[F, Int]): F[Unit] =
+    currentPage
+      .getAndUpdate(_ + 1)
+      .map(query)
+      .flatMap(findRecords[projects.Slug])
+      .map(asToBeMigratedInserts)
+      .flatMap(storeInBacklog(currentPage))
+
+  private def query(page: Int) = SparqlQuery.ofUnsafe(
+    show"${AddProjectSlug.name} - projects to migrate",
+    Prefixes of (renku -> "renku", schema -> "schema"),
+    sparql"""|SELECT DISTINCT ?slug
+             |WHERE {
+             |  GRAPH ?id {
+             |    ?id a schema:Project.
+             |  }
+             |  {
+             |    GRAPH ${GraphClass.Projects.id} {
+             |      ?id renku:projectPath ?slug
+             |      FILTER NOT EXISTS {
+             |        ?id renku:slug ?slg
+             |      }
+             |    }
+             |  } UNION {
+             |    GRAPH ?id {
+             |      ?id renku:projectPath ?slug
+             |      FILTER NOT EXISTS {
+             |        ?id renku:slug ?slg
+             |      }
+             |    }
+             |  }
+             |}
+             |ORDER BY ?slug
+             |LIMIT $pageSize
+             |OFFSET ${(page - 1) * pageSize}
+             |""".stripMargin
+  )
+
+  private implicit lazy val decoder: Decoder[List[projects.Slug]] = ResultsDecoder[List, projects.Slug] {
+    implicit cur => extract[projects.Slug]("slug")
+  }
+
+  private def storeInBacklog(currentPage: Ref[F, Int]): Option[SparqlQuery] => F[Unit] = {
+    case None        => ().pure[F]
+    case Some(query) => migrationsDSClient.updateWithNoResult(query) >> addPageToBacklog(currentPage)
+  }
+}

triples-generator/src/main/scala/io/renku/triplesgenerator/events/consumers/tsmigrationrequest/migrations/projectslug/MigrationNeedChecker.scala

@@ -0,0 +1,84 @@
+/*
+ * Copyright 2023 Swiss Data Science Center (SDSC)
+ * A partnership between École Polytechnique Fédérale de Lausanne (EPFL) and
+ * Eidgenössische Technische Hochschule Zürich (ETHZ).
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.renku.triplesgenerator.events.consumers.tsmigrationrequest
+package migrations.projectslug
+
+import cats.MonadThrow
+import cats.effect.Async
+import cats.syntax.all._
+import eu.timepit.refined.auto._
+import io.circe.Decoder
+import io.renku.graph.model.GraphClass
+import io.renku.graph.model.Schemas._
+import io.renku.triplesstore.SparqlQuery.Prefixes
+import io.renku.triplesstore._
+import io.renku.triplesstore.client.syntax._
+import org.typelevel.log4cats.Logger
+
+private trait MigrationNeedChecker[F[_]] {
+  def checkMigrationNeeded: F[ConditionedMigration.MigrationRequired]
+}
+
+private object MigrationNeedChecker {
+  def apply[F[_]: Async: Logger: SparqlQueryTimeRecorder]: F[MigrationNeedChecker[F]] =
+    ProjectsConnectionConfig[F]().map(TSClient[F](_)).map(new MigrationNeedCheckerImpl[F](_))
+}
+
+private class MigrationNeedCheckerImpl[F[_]: MonadThrow](tsClient: TSClient[F]) extends MigrationNeedChecker[F] {
+
+  override def checkMigrationNeeded: F[ConditionedMigration.MigrationRequired] =
+    tsClient.queryExpecting[Int](query).map {
+      case 0       => ConditionedMigration.MigrationRequired.No("all projects have slug property")
+      case nonZero => ConditionedMigration.MigrationRequired.Yes(s"$nonZero projects does not have slug property")
+    }
+
+  private lazy val query = SparqlQuery.ofUnsafe(
+    show"${AddProjectSlug.name} - check migration needed",
+    Prefixes of (renku -> "renku", schema -> "schema"),
+    sparql"""|SELECT (COUNT(DISTINCT ?id) AS ?cnt)
+             |WHERE {
+             |  GRAPH ?id {
+             |    ?id a schema:Project
+             |  }
+             |  {
+             |    GRAPH ${GraphClass.Projects.id} {
+             |      ?id renku:projectPath ?slug
+             |      FILTER NOT EXISTS {
+             |        ?id renku:slug ?slg
+             |      }
+             |    }
+             |  } UNION {
+             |    GRAPH ?id {
+             |      ?id renku:projectPath ?slug
+             |      FILTER NOT EXISTS {
+             |        ?id renku:slug ?slg
+             |      }
+             |    }
+             |  }
+             |}
+             |LIMIT 1
+             |""".stripMargin
+  )
+
+  import io.renku.triplesstore.ResultsDecoder._
+
+  private implicit lazy val decoder: Decoder[Int] = ResultsDecoder.single[Int] { implicit cur =>
+    extract[String]("cnt").map(_.toInt)
+  }
+}