Add organization deletion route - delete scoped partitions and global…

… data

delta/app/src/main/scala/ch/epfl/bluebrain/nexus/delta/routes/OrganizationsRoutes.scala

@@ -10,6 +10,7 @@ import ch.epfl.bluebrain.nexus.delta.rdf.utils.JsonKeyOrdering
 import ch.epfl.bluebrain.nexus.delta.routes.OrganizationsRoutes.OrganizationInput
 import ch.epfl.bluebrain.nexus.delta.sdk.OrganizationResource
 import ch.epfl.bluebrain.nexus.delta.sdk.acls.AclCheck
+import ch.epfl.bluebrain.nexus.delta.sdk.ce.DeltaDirectives.{emit => emitCE}
 import ch.epfl.bluebrain.nexus.delta.sdk.circe.CirceUnmarshalling
 import ch.epfl.bluebrain.nexus.delta.sdk.directives.DeltaDirectives._
 import ch.epfl.bluebrain.nexus.delta.sdk.directives.{AuthDirectives, DeltaSchemeDirectives}
@@ -20,9 +21,9 @@ import ch.epfl.bluebrain.nexus.delta.sdk.model.BaseUri
 import ch.epfl.bluebrain.nexus.delta.sdk.model.search.SearchParams.OrganizationSearchParams
 import ch.epfl.bluebrain.nexus.delta.sdk.model.search.SearchResults._
 import ch.epfl.bluebrain.nexus.delta.sdk.model.search.{PaginationConfig, SearchResults}
-import ch.epfl.bluebrain.nexus.delta.sdk.organizations.Organizations
 import ch.epfl.bluebrain.nexus.delta.sdk.organizations.model.OrganizationRejection._
 import ch.epfl.bluebrain.nexus.delta.sdk.organizations.model.{Organization, OrganizationRejection}
+import ch.epfl.bluebrain.nexus.delta.sdk.organizations.{OrganizationDeleter, Organizations}
 import ch.epfl.bluebrain.nexus.delta.sdk.permissions.Permissions._
 import io.circe.Decoder
 import io.circe.generic.extras.Configuration
@@ -47,6 +48,7 @@ import scala.annotation.nowarn
 final class OrganizationsRoutes(
     identities: Identities,
     organizations: Organizations,
+    orgDeleter: OrganizationDeleter,
     aclCheck: AclCheck,
     schemeDirectives: DeltaSchemeDirectives
 )(implicit
@@ -113,11 +115,20 @@ final class OrganizationsRoutes(
                       }
                     }
                   },
-                  // Deprecate organization
+                  // Deprecate or delete organization
                   delete {
-                    authorizeFor(id, orgs.write).apply {
-                      parameter("rev".as[Int]) { rev => emit(organizations.deprecate(id, rev).mapValue(_.metadata)) }
-                    }
+                    concat(
+                      parameter("rev".as[Int]) { rev =>
+                        authorizeFor(id, orgs.write).apply {
+                          emit(organizations.deprecate(id, rev).mapValue(_.metadata))
+                        }
+                      },
+                      parameter("prune".requiredValue(true)) { _ =>
+                        authorizeFor(id, orgs.delete).apply {
+                          emitCE(orgDeleter.delete(id).attemptNarrow[OrganizationRejection])
+                        }
+                      }
+                    )
                   }
                 )
               }
@@ -154,6 +165,7 @@ object OrganizationsRoutes {
   def apply(
       identities: Identities,
       organizations: Organizations,
+      orgDeleter: OrganizationDeleter,
       aclCheck: AclCheck,
       schemeDirectives: DeltaSchemeDirectives
   )(implicit
@@ -163,6 +175,6 @@ object OrganizationsRoutes {
       cr: RemoteContextResolution,
       ordering: JsonKeyOrdering
   ): Route =
-    new OrganizationsRoutes(identities, organizations, aclCheck, schemeDirectives).routes
+    new OrganizationsRoutes(identities, organizations, orgDeleter, aclCheck, schemeDirectives).routes
 
 }

delta/app/src/main/scala/ch/epfl/bluebrain/nexus/delta/wiring/DeltaModule.scala

@@ -61,6 +61,8 @@ class DeltaModule(appCfg: AppConfig, config: Config)(implicit classLoader: Class
   make[StrictEntity].from { appCfg.http.strictEntityTimeout }
   make[ServiceAccount].from { appCfg.serviceAccount.value }
 
+  implicit val scheduler: Scheduler = Scheduler.global
+
   make[Transactors].fromResource {
     Transactors.init(appCfg.database)
   }
@@ -102,7 +104,7 @@ class DeltaModule(appCfg: AppConfig, config: Config)(implicit classLoader: Class
   make[Clock[UIO]].from(Clock[UIO])
   make[Clock[IO]].from(Clock.create[IO])
   make[UUIDF].from(UUIDF.random)
-  make[Scheduler].from(Scheduler.global)
+  make[Scheduler].from(scheduler)
   make[JsonKeyOrdering].from(
     JsonKeyOrdering.default(topKeys =
       List("@context", "@id", "@type", "reason", "details", "sourceId", "projectionId", "_total", "_results")

delta/app/src/main/scala/ch/epfl/bluebrain/nexus/delta/wiring/OrganizationsModule.scala

@@ -20,6 +20,7 @@ import ch.epfl.bluebrain.nexus.delta.sourcing.Transactors
 import izumi.distage.model.definition.{Id, ModuleDef}
 import monix.bio.UIO
 import monix.execution.Scheduler
+import ch.epfl.bluebrain.nexus.delta.sdk.organizations.OrganizationDeleter
 
 /**
   * Organizations module wiring config.
@@ -43,18 +44,23 @@ object OrganizationsModule extends ModuleDef {
       )(clock, uuidF)
   }
 
+  make[OrganizationDeleter].from { (xas: Transactors) =>
+    OrganizationDeleter(xas)
+  }
+
   make[OrganizationsRoutes].from {
     (
         identities: Identities,
         organizations: Organizations,
+        orgDeleter: OrganizationDeleter,
         cfg: AppConfig,
         aclCheck: AclCheck,
         schemeDirectives: DeltaSchemeDirectives,
         s: Scheduler,
         cr: RemoteContextResolution @Id("aggregate"),
         ordering: JsonKeyOrdering
     ) =>
-      new OrganizationsRoutes(identities, organizations, aclCheck, schemeDirectives)(
+      new OrganizationsRoutes(identities, organizations, orgDeleter, aclCheck, schemeDirectives)(
         cfg.http.baseUri,
         cfg.organizations.pagination,
         s,

delta/app/src/test/scala/ch/epfl/bluebrain/nexus/delta/routes/OrganizationsRoutesSpec.scala

@@ -3,7 +3,8 @@ package ch.epfl.bluebrain.nexus.delta.routes
 import akka.http.scaladsl.model.StatusCodes
 import akka.http.scaladsl.model.headers.OAuth2BearerToken
 import akka.http.scaladsl.server.Route
-import ch.epfl.bluebrain.nexus.delta.kernel.utils.{UUIDF, UrlUtils}
+import ch.epfl.bluebrain.nexus.delta.kernel.utils.UUIDF
+import ch.epfl.bluebrain.nexus.delta.kernel.utils.UrlUtils
 import ch.epfl.bluebrain.nexus.delta.rdf.Vocabulary.contexts
 import ch.epfl.bluebrain.nexus.delta.sdk.acls.AclSimpleCheck
 import ch.epfl.bluebrain.nexus.delta.sdk.acls.model.AclAddress
@@ -12,14 +13,22 @@ import ch.epfl.bluebrain.nexus.delta.sdk.generators.OrganizationGen
 import ch.epfl.bluebrain.nexus.delta.sdk.identities.IdentitiesDummy
 import ch.epfl.bluebrain.nexus.delta.sdk.identities.model.Caller
 import ch.epfl.bluebrain.nexus.delta.sdk.implicits._
-import ch.epfl.bluebrain.nexus.delta.sdk.organizations.{OrganizationsConfig, OrganizationsImpl}
-import ch.epfl.bluebrain.nexus.delta.sdk.permissions.Permissions.{events, orgs => orgsPermissions}
+import ch.epfl.bluebrain.nexus.delta.sdk.organizations.OrganizationsConfig
+import ch.epfl.bluebrain.nexus.delta.sdk.organizations.OrganizationDeleter
+import ch.epfl.bluebrain.nexus.delta.sdk.organizations.OrganizationsImpl
+import ch.epfl.bluebrain.nexus.delta.sdk.organizations.model.OrganizationRejection.OrganizationNonEmpty
+import ch.epfl.bluebrain.nexus.delta.sdk.permissions.Permissions.events
+import ch.epfl.bluebrain.nexus.delta.sdk.permissions.Permissions.{orgs => orgsPermissions}
 import ch.epfl.bluebrain.nexus.delta.sdk.projects.OwnerPermissionsScopeInitialization
 import ch.epfl.bluebrain.nexus.delta.sdk.utils.BaseRouteSpec
-import ch.epfl.bluebrain.nexus.delta.sourcing.model.Identity.{Anonymous, Authenticated, Group, Subject}
+import ch.epfl.bluebrain.nexus.delta.sourcing.model.Identity.Anonymous
+import ch.epfl.bluebrain.nexus.delta.sourcing.model.Identity.Authenticated
+import ch.epfl.bluebrain.nexus.delta.sourcing.model.Identity.Group
+import ch.epfl.bluebrain.nexus.delta.sourcing.model.Identity.Subject
 import ch.epfl.bluebrain.nexus.delta.sourcing.model.Label
 import ch.epfl.bluebrain.nexus.testkit.bio.IOFromMap
 import io.circe.Json
+import cats.effect.IO
 
 import java.util.UUID
 
@@ -38,7 +47,11 @@ class OrganizationsRoutesSpec extends BaseRouteSpec with IOFromMap {
     aclChecker.append,
     Set(orgsPermissions.write, orgsPermissions.read)
   )
-  private lazy val orgs  = OrganizationsImpl(Set(aopd), config, xas)
+
+  private lazy val orgs                            = OrganizationsImpl(Set(aopd), config, xas)
+  private lazy val orgDeleter: OrganizationDeleter = id =>
+    if (id == org1.label) IO.raiseError(OrganizationNonEmpty(id))
+    else IO.unit
 
   private val caller = Caller(alice, Set(alice, Anonymous, Authenticated(realm), Group("group", realm)))
 
@@ -48,6 +61,7 @@ class OrganizationsRoutesSpec extends BaseRouteSpec with IOFromMap {
     OrganizationsRoutes(
       identities,
       orgs,
+      orgDeleter,
       aclChecker,
       DeltaSchemeDirectives.onlyResolveOrgUuid(ioFromMap(fixedUuid -> org1.label))
     )
@@ -212,6 +226,33 @@ class OrganizationsRoutesSpec extends BaseRouteSpec with IOFromMap {
       }
     }
 
+    "fail to deprecate an organization if the revision is omitted" in {
+      Delete("/v1/orgs/org2") ~> addCredentials(OAuth2BearerToken("alice")) ~> routes ~> check {
+        status shouldEqual StatusCodes.BadRequest
+      }
+    }
+
+    "delete an organization" in {
+      aclChecker.append(AclAddress.fromOrg(org2.label), caller.subject -> Set(orgsPermissions.delete)).accepted
+      Delete("/v1/orgs/org2?prune=true") ~> addCredentials(OAuth2BearerToken("alice")) ~> routes ~> check {
+        status shouldEqual StatusCodes.OK
+      }
+    }
+
+    "fail when trying to delete a non-empty organization" in {
+      aclChecker.append(AclAddress.fromOrg(org1.label), caller.subject -> Set(orgsPermissions.delete)).accepted
+      Delete("/v1/orgs/org1?prune=true") ~> addCredentials(OAuth2BearerToken("alice")) ~> routes ~> check {
+        status shouldEqual StatusCodes.Conflict
+      }
+    }
+
+    "fail to delete an organization without organizations/delete permission" in {
+      aclChecker.subtract(AclAddress.fromOrg(org2.label), caller.subject -> Set(orgsPermissions.delete)).accepted
+      Delete("/v1/orgs/org2?prune=true") ~> addCredentials(OAuth2BearerToken("alice")) ~> routes ~> check {
+        status shouldEqual StatusCodes.Forbidden
+      }
+    }
+
     "fail fetch an organization without organizations/read permission" in {
       aclChecker.delete(Label.unsafe("org1")).accepted
       forAll(

delta/sdk/src/main/scala/ch/epfl/bluebrain/nexus/delta/sdk/acls/Acls.scala

@@ -201,10 +201,10 @@ trait Acls {
   def delete(address: AclAddress, rev: Int)(implicit caller: Subject): IO[AclRejection, AclResource]
 
   /**
-    * Hard deletes events and states for the given acl address This is meant to be used internally for the project
-    * deletion feature
+    * Hard deletes events and states for the given acl address. This is meant to be used internally for project and
+    * organization deletion.
     */
-  def internalDelete(project: AclAddress): UIO[Unit]
+  def purge(project: AclAddress): UIO[Unit]
 
   private def filterSelf(resource: AclResource)(implicit caller: Caller): AclResource =
     resource.map(_.filter(caller.identities))
@@ -363,7 +363,7 @@ object Acls {
   def projectDeletionTask(acls: Acls): ProjectDeletionTask = new ProjectDeletionTask {
     override def apply(project: ProjectRef)(implicit subject: Subject): Task[ProjectDeletionReport.Stage] =
       acls
-        .internalDelete(project)
+        .purge(project)
         .as(
           ProjectDeletionReport.Stage("acls", "The acl has been deleted.")
         )

delta/sdk/src/main/scala/ch/epfl/bluebrain/nexus/delta/sdk/acls/AclsImpl.scala

@@ -89,7 +89,7 @@ final class AclsImpl private (
 
   private def eval(cmd: AclCommand): IO[AclRejection, AclResource] = log.evaluate(cmd.address, cmd).map(_._2.toResource)
 
-  override def internalDelete(project: AclAddress): UIO[Unit] = log.delete(project)
+  override def purge(project: AclAddress): UIO[Unit] = log.delete(project)
 }
 
 object AclsImpl {

delta/sdk/src/main/scala/ch/epfl/bluebrain/nexus/delta/sdk/organizations/OrganizationDeleter.scala

@@ -0,0 +1,61 @@
+package ch.epfl.bluebrain.nexus.delta.sdk.organizations
+
+import cats.effect.IO
+import cats.syntax.all._
+import ch.epfl.bluebrain.nexus.delta.kernel.Logger
+import ch.epfl.bluebrain.nexus.delta.rdf.IriOrBNode
+import ch.epfl.bluebrain.nexus.delta.sdk.acls.Acls
+import ch.epfl.bluebrain.nexus.delta.sdk.acls.model.AclAddress
+import ch.epfl.bluebrain.nexus.delta.sdk.organizations.model.OrganizationRejection.OrganizationNonEmpty
+import ch.epfl.bluebrain.nexus.delta.sourcing.model.{EntityType, Label}
+import ch.epfl.bluebrain.nexus.delta.sourcing.{PartitionInit, Transactors}
+import doobie.implicits._
+import ch.epfl.bluebrain.nexus.delta.sourcing.implicits._
+import doobie.util.update.Update0
+import doobie.{ConnectionIO, Update}
+import org.typelevel.log4cats.{Logger => Log4CatsLogger}
+
+trait OrganizationDeleter {
+  def delete(id: Label): IO[Unit]
+}
+
+object OrganizationDeleter {
+
+  def apply(xas: Transactors): OrganizationDeleter = new OrganizationDeleter {
+
+    private val log: Log4CatsLogger[IO] = Logger.cats[OrganizationDeleter]
+
+    def delete(id: Label): IO[Unit] =
+      for {
+        canDelete <- orgIsEmpty(id)
+        _         <- if (canDelete) log.info(s"Deleting empty organization $id") *> deleteAll(id)
+                     else log.error(s"Failed to delete empty organization $id") *> IO.raiseError(OrganizationNonEmpty(id))
+      } yield ()
+
+    private def deleteAll(id: Label): IO[Unit] =
+      (for {
+        _ <- List("scoped_events", "scoped_states").traverse(dropPartition(id, _))
+        _ <- List("global_events", "global_states").traverse(deleteGlobal(id, _))
+      } yield ()).transact(xas.writeCE).void
+
+    private def dropPartition(id: Label, table: String): ConnectionIO[Unit] =
+      Update0(s"DROP TABLE IF EXISTS ${PartitionInit.orgPartition(table, id)}", None).run.void
+
+    private def deleteGlobal(id: Label, table: String): ConnectionIO[Unit] =
+      for {
+        _ <- deleteGlobalQuery(Organizations.encodeId(id), Organizations.entityType, table)
+        _ <- deleteGlobalQuery(Acls.encodeId(AclAddress.fromOrg(id)), Acls.entityType, table)
+      } yield ()
+
+    private def deleteGlobalQuery(id: IriOrBNode.Iri, tpe: EntityType, table: String): ConnectionIO[Unit] =
+      Update[(EntityType, IriOrBNode.Iri)](s"DELETE FROM $table WHERE" ++ " type = ? AND id = ?").run((tpe, id)).void
+
+    private def orgIsEmpty(id: Label): IO[Boolean] =
+      sql"SELECT type from scoped_events WHERE org = $id LIMIT 1"
+        .query[Label]
+        .option
+        .map(_.isEmpty)
+        .transact(xas.readCE)
+  }
+
+}

delta/sdk/src/main/scala/ch/epfl/bluebrain/nexus/delta/sdk/organizations/Organizations.scala

@@ -195,8 +195,8 @@ object Organizations {
 
     command match {
       case c: CreateOrganization    => create(c)
-      case c: UpdateOrganization    => update(c)
-      case c: DeprecateOrganization => deprecate(c)
+      case u: UpdateOrganization    => update(u)
+      case d: DeprecateOrganization => deprecate(d)
     }
   }
 

delta/sdk/src/main/scala/ch/epfl/bluebrain/nexus/delta/sdk/organizations/OrganizationsImpl.scala

@@ -4,18 +4,24 @@ import cats.effect.Clock
 import ch.epfl.bluebrain.nexus.delta.kernel.kamon.KamonMetricComponent
 import ch.epfl.bluebrain.nexus.delta.kernel.search.Pagination
 import ch.epfl.bluebrain.nexus.delta.kernel.utils.UUIDF
-import ch.epfl.bluebrain.nexus.delta.sdk.model.search.{SearchParams, SearchResults}
+import ch.epfl.bluebrain.nexus.delta.sdk.OrganizationResource
+import ch.epfl.bluebrain.nexus.delta.sdk.ScopeInitialization
+import ch.epfl.bluebrain.nexus.delta.sdk.model.search.SearchParams
+import ch.epfl.bluebrain.nexus.delta.sdk.model.search.SearchResults
 import ch.epfl.bluebrain.nexus.delta.sdk.organizations.Organizations.entityType
 import ch.epfl.bluebrain.nexus.delta.sdk.organizations.OrganizationsImpl.OrganizationsLog
+import ch.epfl.bluebrain.nexus.delta.sdk.organizations.model.OrganizationCommand
 import ch.epfl.bluebrain.nexus.delta.sdk.organizations.model.OrganizationCommand._
+import ch.epfl.bluebrain.nexus.delta.sdk.organizations.model.OrganizationEvent
+import ch.epfl.bluebrain.nexus.delta.sdk.organizations.model.OrganizationRejection
 import ch.epfl.bluebrain.nexus.delta.sdk.organizations.model.OrganizationRejection._
-import ch.epfl.bluebrain.nexus.delta.sdk.organizations.model.{OrganizationCommand, OrganizationEvent, OrganizationRejection, OrganizationState}
+import ch.epfl.bluebrain.nexus.delta.sdk.organizations.model.OrganizationState
 import ch.epfl.bluebrain.nexus.delta.sdk.syntax._
-import ch.epfl.bluebrain.nexus.delta.sdk.{OrganizationResource, ScopeInitialization}
 import ch.epfl.bluebrain.nexus.delta.sourcing._
 import ch.epfl.bluebrain.nexus.delta.sourcing.model.Identity.Subject
 import ch.epfl.bluebrain.nexus.delta.sourcing.model.Label
-import monix.bio.{IO, UIO}
+import monix.bio.IO
+import monix.bio.UIO
 
 final class OrganizationsImpl private (
     log: OrganizationsLog,

delta/sdk/src/main/scala/ch/epfl/bluebrain/nexus/delta/sdk/organizations/model/OrganizationRejection.scala

@@ -18,7 +18,7 @@ import io.circe.{Encoder, JsonObject}
   * @param reason
   *   a descriptive message as to why the rejection occurred
   */
-sealed abstract class OrganizationRejection(val reason: String) extends Product with Serializable
+sealed abstract class OrganizationRejection(val reason: String) extends Throwable with Product with Serializable
 
 object OrganizationRejection {
 
@@ -68,14 +68,23 @@ object OrganizationRejection {
       )
 
   /**
-    * Signals and attempt to update/deprecate an organization that is already deprecated.
+    * Signals an attempt to update/deprecate an organization that is already deprecated.
     *
     * @param label
     *   the label of the organization
     */
   final case class OrganizationIsDeprecated(label: Label)
       extends OrganizationRejection(s"Organization '$label' is deprecated.")
 
+  /**
+    * Signals an attempt to delete an organization that contains at least one project.
+    *
+    * @param label
+    *   the label of the organization
+    */
+  final case class OrganizationNonEmpty(label: Label)
+      extends OrganizationRejection(s"Organization '$label' cannot be deleted since it contains at least one project.")
+
   /**
     * Rejection returned when the organization initialization could not be performed.
     *
@@ -108,6 +117,7 @@ object OrganizationRejection {
       case OrganizationRejection.OrganizationNotFound(_)      => StatusCodes.NotFound
       case OrganizationRejection.OrganizationAlreadyExists(_) => StatusCodes.Conflict
       case OrganizationRejection.IncorrectRev(_, _)           => StatusCodes.Conflict
+      case OrganizationRejection.OrganizationNonEmpty(_)      => StatusCodes.Conflict
       case OrganizationRejection.RevisionNotFound(_, _)       => StatusCodes.NotFound
       case _                                                  => StatusCodes.BadRequest
     }

delta/sdk/src/main/scala/ch/epfl/bluebrain/nexus/delta/sdk/permissions/Permissions.scala

@@ -157,6 +157,8 @@ object Permissions {
     final val read: Permission   = Permission.unsafe("organizations/read")
     final val write: Permission  = Permission.unsafe("organizations/write")
     final val create: Permission = Permission.unsafe("organizations/create")
+    // TODO setup this permission
+    final val delete: Permission = Permission.unsafe("organizations/delete")
   }
 
   /**