Repository: BlueBrain/nexus (scala)

Issue:
Export selected scoped events in a file
**Motivation**
In order to share resources and projects between environments, we need to be able to export those events

**Acceptance criteria**
1. A user provides a list of projects to export
2. A user can provide a starting offset to do a partial export
3. An endpoint is provided to trigger the export with this parameter
4. This endpoint is protected by a custom permission
5. The rows are exported with all existing colums as json in a file in a configured directory where each row takes one line
6. The rows are sorted by ascending offset

Example on how to fetch the events:
```
val q =
     """
      copy (select row_to_json(t) from (select * from scoped_events) t)
      to stdout (
        encoding 'utf-8',
        format text
      )
      """

    PHC.pgGetCopyAPI(PFCM.copyOut(q, Console.out)).transact(xas.streaming).accepted
```

**Not in scope**
1. The import part
2. The export of physical files

Pull request:
Fixes #4662

Patch:
diff --git a/delta/app/src/main/resources/app.conf b/delta/app/src/main/resources/app.conf
index fb910fc435..e479a3feaa 100644
--- a/delta/app/src/main/resources/app.conf
+++ b/delta/app/src/main/resources/app.conf
@@ -43,6 +43,14 @@ app {
     }
   }
 
+  # Database export configuration
+  export {
+    # Max number of concurrent exports
+    permits = 1
+    # Target directory for exports
+    target = "/tmp"
+  }
+
   # Fusion configuration
   fusion {
     # The base url for Fusion
@@ -116,7 +124,8 @@ app {
       "storages/write",
       "version/read",
       "quotas/read",
-      "supervision/read"
+      "supervision/read",
+      "export/run"
     ]
 
     # permissions applied to the creator of the project
diff --git a/delta/app/src/main/scala/ch/epfl/bluebrain/nexus/delta/config/AppConfig.scala b/delta/app/src/main/scala/ch/epfl/bluebrain/nexus/delta/config/AppConfig.scala
index 424705c65d..c141ff29fd 100644
--- a/delta/app/src/main/scala/ch/epfl/bluebrain/nexus/delta/config/AppConfig.scala
+++ b/delta/app/src/main/scala/ch/epfl/bluebrain/nexus/delta/config/AppConfig.scala
@@ -18,6 +18,7 @@ import ch.epfl.bluebrain.nexus.delta.sdk.resources.ResourcesConfig
 import ch.epfl.bluebrain.nexus.delta.sdk.schemas.SchemasConfig
 import ch.epfl.bluebrain.nexus.delta.sdk.sse.SseConfig
 import ch.epfl.bluebrain.nexus.delta.sourcing.config.{DatabaseConfig, ProjectionConfig}
+import ch.epfl.bluebrain.nexus.delta.sourcing.exporter.ExportConfig
 import com.typesafe.config.{Config, ConfigFactory, ConfigParseOptions, ConfigResolveOptions}
 import pureconfig.generic.semiauto.deriveReader
 import pureconfig.{ConfigReader, ConfigSource}
@@ -47,7 +48,8 @@ final case class AppConfig(
     serviceAccount: ServiceAccountConfig,
     sse: SseConfig,
     projections: ProjectionConfig,
-    fusion: FusionConfig
+    fusion: FusionConfig,
+    `export`: ExportConfig
 )
 
 object AppConfig {
diff --git a/delta/app/src/main/scala/ch/epfl/bluebrain/nexus/delta/routes/ExportRoutes.scala b/delta/app/src/main/scala/ch/epfl/bluebrain/nexus/delta/routes/ExportRoutes.scala
new file mode 100644
index 0000000000..f1a644da1e
--- /dev/null
+++ b/delta/app/src/main/scala/ch/epfl/bluebrain/nexus/delta/routes/ExportRoutes.scala
@@ -0,0 +1,40 @@
+package ch.epfl.bluebrain.nexus.delta.routes
+
+import akka.http.scaladsl.model.StatusCodes
+import akka.http.scaladsl.server.Route
+import ch.epfl.bluebrain.nexus.delta.rdf.jsonld.context.RemoteContextResolution
+import ch.epfl.bluebrain.nexus.delta.rdf.utils.JsonKeyOrdering
+import ch.epfl.bluebrain.nexus.delta.sdk.acls.AclCheck
+import ch.epfl.bluebrain.nexus.delta.sdk.acls.model.AclAddress
+import ch.epfl.bluebrain.nexus.delta.sdk.circe.CirceUnmarshalling
+import ch.epfl.bluebrain.nexus.delta.sdk.directives.AuthDirectives
+import ch.epfl.bluebrain.nexus.delta.sdk.directives.DeltaDirectives._
+import ch.epfl.bluebrain.nexus.delta.sdk.directives.UriDirectives.baseUriPrefix
+import ch.epfl.bluebrain.nexus.delta.sdk.identities.Identities
+import ch.epfl.bluebrain.nexus.delta.sdk.model.BaseUri
+import ch.epfl.bluebrain.nexus.delta.sdk.permissions.Permissions
+import ch.epfl.bluebrain.nexus.delta.sourcing.exporter.{ExportEventQuery, Exporter}
+
+class ExportRoutes(identities: Identities, aclCheck: AclCheck, exporter: Exporter)(implicit
+    baseUri: BaseUri,
+    cr: RemoteContextResolution,
+    ordering: JsonKeyOrdering
+) extends AuthDirectives(identities, aclCheck)
+    with CirceUnmarshalling {
+
+  def routes: Route =
+    baseUriPrefix(baseUri.prefix) {
+      pathPrefix("export") {
+        pathPrefix("events") {
+          extractCaller { implicit caller =>
+            (post & pathEndOrSingleSlash & entity(as[ExportEventQuery])) { query =>
+              authorizeFor(AclAddress.Root, Permissions.exporter.run).apply {
+                emit(StatusCodes.Accepted, exporter.events(query).start.void)
+              }
+            }
+          }
+        }
+      }
+    }
+
+}
diff --git a/delta/app/src/main/scala/ch/epfl/bluebrain/nexus/delta/wiring/ExportModule.scala b/delta/app/src/main/scala/ch/epfl/bluebrain/nexus/delta/wiring/ExportModule.scala
new file mode 100644
index 0000000000..0af4dc0e07
--- /dev/null
+++ b/delta/app/src/main/scala/ch/epfl/bluebrain/nexus/delta/wiring/ExportModule.scala
@@ -0,0 +1,44 @@
+package ch.epfl.bluebrain.nexus.delta.wiring
+
+import cats.effect.{Clock, IO}
+import ch.epfl.bluebrain.nexus.delta.Main.pluginsMaxPriority
+import ch.epfl.bluebrain.nexus.delta.config.AppConfig
+import ch.epfl.bluebrain.nexus.delta.rdf.jsonld.context.RemoteContextResolution
+import ch.epfl.bluebrain.nexus.delta.rdf.utils.JsonKeyOrdering
+import ch.epfl.bluebrain.nexus.delta.routes.ExportRoutes
+import ch.epfl.bluebrain.nexus.delta.sdk.PriorityRoute
+import ch.epfl.bluebrain.nexus.delta.sdk.acls.AclCheck
+import ch.epfl.bluebrain.nexus.delta.sdk.identities.Identities
+import ch.epfl.bluebrain.nexus.delta.sourcing.Transactors
+import ch.epfl.bluebrain.nexus.delta.sourcing.exporter.Exporter
+import izumi.distage.model.definition.{Id, ModuleDef}
+
+// $COVERAGE-OFF$
+object ExportModule extends ModuleDef {
+
+  make[Exporter].fromEffect { (config: AppConfig, clock: Clock[IO], xas: Transactors) =>
+    Exporter(config.`export`, clock, xas)
+  }
+
+  make[ExportRoutes].from {
+    (
+        cfg: AppConfig,
+        identities: Identities,
+        aclCheck: AclCheck,
+        exporter: Exporter,
+        cr: RemoteContextResolution @Id("aggregate"),
+        ordering: JsonKeyOrdering
+    ) =>
+      new ExportRoutes(identities, aclCheck, exporter)(
+        cfg.http.baseUri,
+        cr,
+        ordering
+      )
+  }
+
+  many[PriorityRoute].add { (route: ExportRoutes) =>
+    PriorityRoute(pluginsMaxPriority + 1, route.routes, requiresStrictEntity = true)
+  }
+
+}
+// $COVERAGE-ON$
diff --git a/delta/sdk/src/main/scala/ch/epfl/bluebrain/nexus/delta/sdk/permissions/Permissions.scala b/delta/sdk/src/main/scala/ch/epfl/bluebrain/nexus/delta/sdk/permissions/Permissions.scala
index bc3f036828..e25717590c 100644
--- a/delta/sdk/src/main/scala/ch/epfl/bluebrain/nexus/delta/sdk/permissions/Permissions.scala
+++ b/delta/sdk/src/main/scala/ch/epfl/bluebrain/nexus/delta/sdk/permissions/Permissions.scala
@@ -176,6 +176,10 @@ object Permissions {
     final val read: Permission = Permission.unsafe("events/read")
   }
 
+  object exporter {
+    final val run: Permission = Permission.unsafe("export/run")
+  }
+
   /**
     * Generic version permissions.
     */
diff --git a/delta/sourcing-psql/src/main/scala/ch/epfl/bluebrain/nexus/delta/sourcing/exporter/ExportConfig.scala b/delta/sourcing-psql/src/main/scala/ch/epfl/bluebrain/nexus/delta/sourcing/exporter/ExportConfig.scala
new file mode 100644
index 0000000000..9c3dcb8914
--- /dev/null
+++ b/delta/sourcing-psql/src/main/scala/ch/epfl/bluebrain/nexus/delta/sourcing/exporter/ExportConfig.scala
@@ -0,0 +1,20 @@
+package ch.epfl.bluebrain.nexus.delta.sourcing.exporter
+
+import fs2.io.file.Path
+import pureconfig.ConfigConvert.catchReadError
+import pureconfig.{ConfigConvert, ConfigReader}
+import pureconfig.generic.semiauto.deriveReader
+
+import scala.annotation.nowarn
+
+final case class ExportConfig(permits: Int, target: Path)
+
+object ExportConfig {
+
+  @nowarn("cat=unused")
+  implicit final val databaseConfigReader: ConfigReader[ExportConfig] = {
+    implicit val pathConverter: ConfigReader[Path] = ConfigConvert.viaString(catchReadError(s => Path(s)), _.toString)
+    deriveReader[ExportConfig]
+  }
+
+}
diff --git a/delta/sourcing-psql/src/main/scala/ch/epfl/bluebrain/nexus/delta/sourcing/exporter/ExportEventQuery.scala b/delta/sourcing-psql/src/main/scala/ch/epfl/bluebrain/nexus/delta/sourcing/exporter/ExportEventQuery.scala
new file mode 100644
index 0000000000..25581ab912
--- /dev/null
+++ b/delta/sourcing-psql/src/main/scala/ch/epfl/bluebrain/nexus/delta/sourcing/exporter/ExportEventQuery.scala
@@ -0,0 +1,19 @@
+package ch.epfl.bluebrain.nexus.delta.sourcing.exporter
+
+import cats.data.NonEmptyList
+import ch.epfl.bluebrain.nexus.delta.sourcing.model.{Label, ProjectRef}
+import ch.epfl.bluebrain.nexus.delta.sourcing.offset.Offset
+import io.circe.Codec
+import io.circe.generic.extras.Configuration
+import io.circe.generic.extras.semiauto.deriveConfiguredCodec
+
+import scala.annotation.nowarn
+
+final case class ExportEventQuery(output: Label, projects: NonEmptyList[ProjectRef], offset: Offset)
+
+object ExportEventQuery {
+
+  @nowarn("cat=unused")
+  implicit private val config: Configuration             = Configuration.default.withStrictDecoding
+  implicit val exportQueryCodec: Codec[ExportEventQuery] = deriveConfiguredCodec[ExportEventQuery]
+}
diff --git a/delta/sourcing-psql/src/main/scala/ch/epfl/bluebrain/nexus/delta/sourcing/exporter/Exporter.scala b/delta/sourcing-psql/src/main/scala/ch/epfl/bluebrain/nexus/delta/sourcing/exporter/Exporter.scala
new file mode 100644
index 0000000000..0c000695ea
--- /dev/null
+++ b/delta/sourcing-psql/src/main/scala/ch/epfl/bluebrain/nexus/delta/sourcing/exporter/Exporter.scala
@@ -0,0 +1,81 @@
+package ch.epfl.bluebrain.nexus.delta.sourcing.exporter
+
+import cats.effect.IO
+import cats.effect.kernel.Clock
+import cats.effect.std.Semaphore
+import ch.epfl.bluebrain.nexus.delta.kernel.Logger
+import ch.epfl.bluebrain.nexus.delta.sourcing.Transactors
+import ch.epfl.bluebrain.nexus.delta.sourcing.exporter.Exporter.ExportResult
+import ch.epfl.bluebrain.nexus.delta.sourcing.implicits._
+import doobie.Fragments
+import doobie.implicits._
+import doobie.util.fragment.Fragment
+import fs2.io.file.{Files, Path}
+import fs2.Stream
+import io.circe.syntax.EncoderOps
+
+import java.time.Instant
+
+trait Exporter {
+
+  def events(query: ExportEventQuery): IO[ExportResult]
+
+}
+
+object Exporter {
+
+  private val logger = Logger[Exporter]
+
+  final case class ExportResult(json: Path, success: Path, start: Instant, end: Instant)
+
+  def apply(config: ExportConfig, clock: Clock[IO], xas: Transactors): IO[Exporter] =
+    Semaphore[IO](config.permits.toLong).map(new ExporterImpl(config.target, _, clock, xas))
+
+  private class ExporterImpl(rootDirectory: Path, semaphore: Semaphore[IO], clock: Clock[IO], xas: Transactors)
+      extends Exporter {
+    override def events(query: ExportEventQuery): IO[ExportResult] = {
+      val projectFilter = Fragments.orOpt(
+        query.projects.map { project => sql"(org = ${project.organization} and project = ${project.project})" }
+      )
+      val q             = asJson(sql"""SELECT *
+             |FROM public.scoped_events
+             |${Fragments.whereAndOpt(projectFilter, query.offset.asFragment)}
+             |ORDER BY ordering
+             |""".stripMargin)
+
+      val exportIO = for {
+        start          <- clock.realTimeInstant
+        _              <- logger.info(s"Starting export for projects ${query.projects} from offset ${query.offset}")
+        targetDirectory = rootDirectory / query.output.value
+        _              <- Files[IO].createDirectory(targetDirectory)
+        exportFile      = targetDirectory / s"$start.json"
+        _              <- exportToFile(q, exportFile)
+        end            <- clock.realTimeInstant
+        exportSuccess   = targetDirectory / s"$start.success"
+        _              <- writeSuccessFile(query, exportSuccess)
+        _              <-
+          logger.info(
+            s"Export for projects ${query.projects} from offset' ${query.offset}' after ${end.getEpochSecond - start.getEpochSecond} seconds."
+          )
+      } yield ExportResult(exportFile, exportSuccess, start, end)
+
+      semaphore.permit.use { _ => exportIO }
+    }
+
+    private def exportToFile(q: Fragment, targetFile: Path) =
+      q.query[String]
+        .stream
+        .intersperse("\n")
+        .transact(xas.streaming)
+        .through(Files[IO].writeUtf8(targetFile))
+        .compile
+        .drain
+
+    private def writeSuccessFile(query: ExportEventQuery, targetFile: Path) =
+      Stream(query.asJson.toString()).through(Files[IO].writeUtf8(targetFile)).compile.drain
+
+    private def asJson(query: Fragment) =
+      sql"""(select row_to_json(t) from ($query) t)"""
+  }
+
+}
diff --git a/storage/src/main/scala/ch/epfl/bluebrain/nexus/storage/config/Settings.scala b/storage/src/main/scala/ch/epfl/bluebrain/nexus/storage/config/Settings.scala
index 5bf49ba40d..28c40e9150 100644
--- a/storage/src/main/scala/ch/epfl/bluebrain/nexus/storage/config/Settings.scala
+++ b/storage/src/main/scala/ch/epfl/bluebrain/nexus/storage/config/Settings.scala
@@ -7,7 +7,6 @@ import pureconfig.ConvertHelpers._
 import pureconfig._
 import pureconfig.generic.auto._
 
-import java.nio.file.{Path, Paths}
 import scala.annotation.nowarn
 
 /**
@@ -22,10 +21,8 @@ class Settings(config: Config) extends Extension {
 
   @nowarn("cat=unused")
   val appConfig: AppConfig = {
-    implicit val uriConverter: ConfigConvert[Uri]   =
+    implicit val uriConverter: ConfigConvert[Uri] =
       ConfigConvert.viaString[Uri](catchReadError(s => Uri(s)), _.toString)
-    implicit val pathConverter: ConfigConvert[Path] =
-      ConfigConvert.viaString[Path](catchReadError(s => Paths.get(s)), _.toString)
     ConfigSource.fromConfig(config).at("app").loadOrThrow[AppConfig]
   }
 


Test patch:
diff --git a/delta/app/src/test/scala/ch/epfl/bluebrain/nexus/delta/routes/ExportRoutesSpec.scala b/delta/app/src/test/scala/ch/epfl/bluebrain/nexus/delta/routes/ExportRoutesSpec.scala
new file mode 100644
index 0000000000..e56b715912
--- /dev/null
+++ b/delta/app/src/test/scala/ch/epfl/bluebrain/nexus/delta/routes/ExportRoutesSpec.scala
@@ -0,0 +1,63 @@
+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 cats.effect.{IO, Ref}
+import ch.epfl.bluebrain.nexus.delta.sdk.acls.AclSimpleCheck
+import ch.epfl.bluebrain.nexus.delta.sdk.acls.model.AclAddress.Root
+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.permissions.Permissions
+import ch.epfl.bluebrain.nexus.delta.sdk.utils.BaseRouteSpec
+import ch.epfl.bluebrain.nexus.delta.sourcing.exporter.Exporter.ExportResult
+import ch.epfl.bluebrain.nexus.delta.sourcing.exporter.{ExportEventQuery, Exporter}
+import ch.epfl.bluebrain.nexus.delta.sourcing.model.Identity.{Anonymous, Authenticated, Group}
+import fs2.io.file.Path
+
+import java.time.Instant
+
+class ExportRoutesSpec extends BaseRouteSpec {
+
+  private val caller = Caller(alice, Set(alice, Anonymous, Authenticated(realm), Group("group", realm)))
+
+  private val identities = IdentitiesDummy(caller)
+
+  private val asAlice = addCredentials(OAuth2BearerToken("alice"))
+
+  private val exportTrigger = Ref.unsafe[IO, Boolean](false)
+
+  private val aclCheck = AclSimpleCheck((alice, Root, Set(Permissions.exporter.run))).accepted
+
+  private val exporter = new Exporter {
+    override def events(query: ExportEventQuery): IO[ExportResult] =
+      exportTrigger.set(true).as(ExportResult(Path("json"), Path("Success"), Instant.EPOCH, Instant.EPOCH))
+  }
+
+  private lazy val routes = Route.seal(
+    new ExportRoutes(
+      identities,
+      aclCheck,
+      exporter
+    ).routes
+  )
+
+  "The export route" should {
+    val query =
+      json"""{ "output": "export-test", "projects": ["org/proj", "org/proj2"], "offset": {"@type": "At", "value": 2}  }"""
+    "fail triggering the export the 'export/run' permission" in {
+      Post("/v1/export/events", query.toEntity) ~> routes ~> check {
+        response.shouldBeForbidden
+        exportTrigger.get.accepted shouldEqual false
+      }
+    }
+
+    "trigger the 'export/run' permission" in {
+      Post("/v1/export/events", query.toEntity) ~> asAlice ~> routes ~> check {
+        response.status shouldEqual StatusCodes.Accepted
+        exportTrigger.get.accepted shouldEqual true
+      }
+    }
+  }
+
+}
diff --git a/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/PullRequest.scala b/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/PullRequest.scala
index e58c6dfff4..d8a6db3880 100644
--- a/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/PullRequest.scala
+++ b/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/PullRequest.scala
@@ -1,6 +1,7 @@
 package ch.epfl.bluebrain.nexus.delta.sourcing
 
 import cats.effect.IO
+import cats.syntax.all._
 import ch.epfl.bluebrain.nexus.delta.kernel.error.ThrowableValue
 import ch.epfl.bluebrain.nexus.delta.rdf.IriOrBNode.Iri
 import ch.epfl.bluebrain.nexus.delta.rdf.Vocabulary.{nxv, rdfs, schemas}
@@ -11,23 +12,39 @@ import ch.epfl.bluebrain.nexus.delta.sourcing.PullRequest.PullRequestCommand._
 import ch.epfl.bluebrain.nexus.delta.sourcing.PullRequest.PullRequestEvent.{PullRequestCreated, PullRequestMerged, PullRequestTagged, PullRequestUpdated}
 import ch.epfl.bluebrain.nexus.delta.sourcing.PullRequest.PullRequestRejection.{AlreadyExists, NotFound, PullRequestAlreadyClosed}
 import ch.epfl.bluebrain.nexus.delta.sourcing.PullRequest.PullRequestState.{PullRequestActive, PullRequestClosed}
+import ch.epfl.bluebrain.nexus.delta.sourcing.config.QueryConfig
 import ch.epfl.bluebrain.nexus.delta.sourcing.event.Event.ScopedEvent
+import ch.epfl.bluebrain.nexus.delta.sourcing.event.ScopedEventStore
 import ch.epfl.bluebrain.nexus.delta.sourcing.model.Identity.{Anonymous, Subject}
 import ch.epfl.bluebrain.nexus.delta.sourcing.model.ResourceRef.Latest
 import ch.epfl.bluebrain.nexus.delta.sourcing.model._
+import ch.epfl.bluebrain.nexus.delta.sourcing.query.RefreshStrategy
 import ch.epfl.bluebrain.nexus.delta.sourcing.state.GraphResource
 import ch.epfl.bluebrain.nexus.delta.sourcing.state.State.ScopedState
 import io.circe.generic.extras.Configuration
 import io.circe.generic.extras.semiauto.deriveConfiguredCodec
 import io.circe.{Codec, Json}
+import doobie.implicits._
 
 import java.time.Instant
 import scala.annotation.nowarn
 
 object PullRequest {
 
+  type EventStore = ScopedEventStore[Iri, PullRequestEvent]
+
   val entityType: EntityType = EntityType("merge-request")
 
+  def eventStore(xas: Transactors, populateWith: PullRequestEvent*): IO[EventStore] = {
+    val store = ScopedEventStore[Iri, PullRequestEvent](
+      PullRequest.entityType,
+      PullRequestEvent.serializer,
+      QueryConfig(10, RefreshStrategy.Stop),
+      xas
+    )
+    populateWith.traverse(store.unsafeSave).transact(xas.write).as(store)
+  }
+
   val stateMachine: StateMachine[PullRequestState, PullRequestCommand, PullRequestEvent] =
     StateMachine(
       None,
diff --git a/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/event/ScopedEventStoreSuite.scala b/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/event/ScopedEventStoreSuite.scala
index 1b1848fc19..5a8f79ea54 100644
--- a/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/event/ScopedEventStoreSuite.scala
+++ b/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/event/ScopedEventStoreSuite.scala
@@ -1,35 +1,22 @@
 package ch.epfl.bluebrain.nexus.delta.sourcing.event
 
-import cats.syntax.all._
-import ch.epfl.bluebrain.nexus.delta.rdf.IriOrBNode.Iri
 import ch.epfl.bluebrain.nexus.delta.rdf.Vocabulary.nxv
 import ch.epfl.bluebrain.nexus.delta.sourcing.PullRequest
-import ch.epfl.bluebrain.nexus.delta.sourcing.PullRequest.PullRequestEvent
 import ch.epfl.bluebrain.nexus.delta.sourcing.PullRequest.PullRequestEvent.{PullRequestCreated, PullRequestMerged, PullRequestUpdated}
-import ch.epfl.bluebrain.nexus.delta.sourcing.config.QueryConfig
 import ch.epfl.bluebrain.nexus.delta.sourcing.model.Identity.{Anonymous, User}
 import ch.epfl.bluebrain.nexus.delta.sourcing.model.{Label, ProjectRef}
 import ch.epfl.bluebrain.nexus.delta.sourcing.postgres.Doobie
-import ch.epfl.bluebrain.nexus.delta.sourcing.query.RefreshStrategy
 import ch.epfl.bluebrain.nexus.testkit.mu.NexusSuite
 import doobie.implicits._
 import munit.AnyFixture
 
 import java.time.Instant
-import scala.concurrent.duration._
 
 class ScopedEventStoreSuite extends NexusSuite with Doobie.Fixture with Doobie.Assertions {
 
-  override def munitFixtures: Seq[AnyFixture[_]] = List(doobie)
+  lazy val fixture = doobieInject(PullRequest.eventStore(_, event1, event2, event3, event4, event5, event6))
 
-  private lazy val xas = doobie()
-
-  private lazy val store = ScopedEventStore[Iri, PullRequestEvent](
-    PullRequest.entityType,
-    PullRequestEvent.serializer,
-    QueryConfig(10, RefreshStrategy.Delay(500.millis)),
-    xas
-  )
+  override def munitFixtures: Seq[AnyFixture[_]] = List(fixture)
 
   private val project1 = ProjectRef.unsafe("org", "proj1")
   private val project2 = ProjectRef.unsafe("org", "proj2")
@@ -49,14 +36,9 @@ class ScopedEventStoreSuite extends NexusSuite with Doobie.Fixture with Doobie.A
 
   private val event6 = PullRequestCreated(id3, project3, Instant.EPOCH, Anonymous)
 
-  private def assertCount = sql"select count(*) from scoped_events".query[Int].unique.transact(xas.read).assertEquals(6)
+  private lazy val (xas, store) = fixture()
 
-  test("Save events") {
-    for {
-      _ <- List(event1, event2, event3, event4, event5, event6).traverse(store.unsafeSave).transact(xas.write)
-      _ <- assertCount
-    } yield ()
-  }
+  private def assertCount = sql"select count(*) from scoped_events".query[Int].unique.transact(xas.read).assertEquals(6)
 
   test("Fail when the PK already exists") {
     for {
diff --git a/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/exporter/ExporterSuite.scala b/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/exporter/ExporterSuite.scala
new file mode 100644
index 0000000000..582a3b19f5
--- /dev/null
+++ b/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/exporter/ExporterSuite.scala
@@ -0,0 +1,99 @@
+package ch.epfl.bluebrain.nexus.delta.sourcing.exporter
+
+import cats.data.NonEmptyList
+import cats.syntax.all._
+import cats.effect.IO
+import ch.epfl.bluebrain.nexus.delta.rdf.Vocabulary.nxv
+import ch.epfl.bluebrain.nexus.delta.sourcing.PullRequest
+import ch.epfl.bluebrain.nexus.delta.sourcing.PullRequest.PullRequestEvent.{PullRequestCreated, PullRequestMerged, PullRequestUpdated}
+import ch.epfl.bluebrain.nexus.delta.sourcing.model.Identity.{Anonymous, User}
+import ch.epfl.bluebrain.nexus.delta.sourcing.model.{Label, ProjectRef}
+import ch.epfl.bluebrain.nexus.delta.sourcing.offset.Offset
+import ch.epfl.bluebrain.nexus.delta.sourcing.postgres.Doobie
+import ch.epfl.bluebrain.nexus.testkit.clock.FixedClock
+import ch.epfl.bluebrain.nexus.testkit.file.TempDirectory
+import ch.epfl.bluebrain.nexus.testkit.mu.NexusSuite
+import fs2.io.file.{Files, Path}
+import io.circe.{parser, DecodingFailure, JsonObject}
+import munit.{AnyFixture, Location}
+
+import java.time.Instant
+
+class ExporterSuite extends NexusSuite with Doobie.Fixture with TempDirectory.Fixture with FixedClock {
+
+  private lazy val doobieFixture                 = doobieInject(
+    PullRequest.eventStore(_, event1, event2, event3, event4, event5, event6),
+    Exporter(exporterConfig, clock, _)
+  )
+  override def munitFixtures: Seq[AnyFixture[_]] = List(tempDirectory, doobieFixture)
+
+  private lazy val exporterConfig   = ExportConfig(3, exportDirectory)
+  private lazy val (_, _, exporter) = doobieFixture()
+  private lazy val exportDirectory  = tempDirectory()
+
+  private val project1 = ProjectRef.unsafe("org", "proj1")
+  private val project2 = ProjectRef.unsafe("org", "proj2")
+  private val project3 = ProjectRef.unsafe("org2", "proj3")
+
+  private val id1 = nxv + "1"
+  private val id2 = nxv + "2"
+  private val id3 = nxv + "3"
+
+  private val event1 = PullRequestCreated(id1, project1, Instant.EPOCH, Anonymous)
+  private val event2 = PullRequestUpdated(id1, project1, 2, Instant.EPOCH, User("Alice", Label.unsafe("Wonderland")))
+  private val event3 = PullRequestMerged(id1, project1, 3, Instant.EPOCH, User("Alice", Label.unsafe("Wonderland")))
+
+  private val event4 = PullRequestCreated(id2, project1, Instant.EPOCH, Anonymous)
+
+  private val event5 = PullRequestCreated(id1, project2, Instant.EPOCH, Anonymous)
+
+  private val event6 = PullRequestCreated(id3, project3, Instant.EPOCH, Anonymous)
+
+  private def parseAsObject(value: String) =
+    parser.parse(value).flatMap(_.asObject.toRight(DecodingFailure("Expected a json object", List.empty)))
+
+  private def orderingValue(obj: JsonObject) =
+    obj("ordering").flatMap(_.asNumber.flatMap(_.toInt))
+
+  private def readJsonLines(path: Path) = Files[IO]
+    .readUtf8Lines(path)
+    .evalMap { line =>
+      IO.fromEither(parseAsObject(line))
+    }
+    .compile
+    .toList
+
+  private def readSuccess(path: Path) = Files[IO]
+    .readUtf8(path)
+    .evalMap { content =>
+      IO.fromEither(parser.decode[ExportEventQuery](content))
+    }
+    .compile
+    .lastOrError
+
+  private def assertExport(result: Exporter.ExportResult, query: ExportEventQuery, expectedOrdering: List[Int])(implicit
+      location: Location
+  ) =
+    for {
+      exportContent <- readJsonLines(result.json)
+      orderingValues = exportContent.mapFilter(orderingValue)
+      _              = assertEquals(orderingValues, expectedOrdering)
+      _             <- readSuccess(result.success).assertEquals(query)
+    } yield ()
+
+  test(s"Export all events for $project1 and $project3") {
+    val query = ExportEventQuery(Label.unsafe("export1"), NonEmptyList.of(project1, project3), Offset.start)
+    for {
+      result <- exporter.events(query)
+      _      <- assertExport(result, query, List(1, 2, 3, 4, 6))
+    } yield ()
+  }
+
+  test(s"Export all events for $project1 and $project3 from offset 3") {
+    val query = ExportEventQuery(Label.unsafe("export2"), NonEmptyList.of(project1, project3), Offset.at(2L))
+    for {
+      result <- exporter.events(query)
+      _      <- assertExport(result, query, List(3, 4, 6))
+    } yield ()
+  }
+}
diff --git a/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/postgres/Doobie.scala b/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/postgres/Doobie.scala
index 95de03f463..d910e7924b 100644
--- a/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/postgres/Doobie.scala
+++ b/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/postgres/Doobie.scala
@@ -8,6 +8,7 @@ import ch.epfl.bluebrain.nexus.testkit.mu.NexusSuite
 import ch.epfl.bluebrain.nexus.testkit.postgres.PostgresContainer
 import doobie.implicits._
 import doobie.postgres.sqlstate
+import munit.catseffect.IOFixture
 import munit.{catseffect, Location}
 import org.postgresql.util.PSQLException
 
@@ -36,6 +37,25 @@ object Doobie {
     val doobie: catseffect.IOFixture[Transactors] =
       ResourceSuiteLocalFixture("doobie", resource(PostgresUser, PostgresPassword))
 
+    def doobieInject[A](f: Transactors => IO[A]): IOFixture[(Transactors, A)] =
+      ResourceSuiteLocalFixture(
+        s"doobie",
+        resource().evalMap { xas =>
+          f(xas).map(xas -> _)
+        }
+      )
+
+    def doobieInject[A, B](f1: Transactors => IO[A], f2: Transactors => IO[B]): IOFixture[(Transactors, A, B)] =
+      ResourceSuiteLocalFixture(
+        s"doobie",
+        resource().evalMap { xas =>
+          for {
+            a <- f1(xas)
+            b <- f2(xas)
+          } yield (xas, a, b)
+        }
+      )
+
     /**
       * Init the partition in the events and states table for the given projects
       */
diff --git a/delta/testkit/src/main/scala/ch/epfl/bluebrain/nexus/testkit/file/TempDirectory.scala b/delta/testkit/src/main/scala/ch/epfl/bluebrain/nexus/testkit/file/TempDirectory.scala
new file mode 100644
index 0000000000..ed78fa59cf
--- /dev/null
+++ b/delta/testkit/src/main/scala/ch/epfl/bluebrain/nexus/testkit/file/TempDirectory.scala
@@ -0,0 +1,19 @@
+package ch.epfl.bluebrain.nexus.testkit.file
+
+import cats.effect.IO
+import cats.effect.kernel.Resource
+import ch.epfl.bluebrain.nexus.testkit.mu.NexusSuite
+import fs2.io.file.{Files, Path}
+import munit.catseffect.IOFixture
+
+object TempDirectory {
+
+  trait Fixture { self: NexusSuite =>
+    val tempDirectory: IOFixture[Path] =
+      ResourceSuiteLocalFixture(
+        "tempDirectory",
+        Resource.make(Files[IO].createTempDirectory)(Files[IO].deleteRecursively)
+      )
+  }
+
+}
diff --git a/tests/src/test/scala/ch/epfl/bluebrain/nexus/tests/iam/types/AclListing.scala b/tests/src/test/scala/ch/epfl/bluebrain/nexus/tests/iam/types/AclListing.scala
index 5429e8d51f..babe14c704 100644
--- a/tests/src/test/scala/ch/epfl/bluebrain/nexus/tests/iam/types/AclListing.scala
+++ b/tests/src/test/scala/ch/epfl/bluebrain/nexus/tests/iam/types/AclListing.scala
@@ -157,6 +157,13 @@ object Permission {
     val list: List[Permission] = Read :: Nil
   }
 
+  object Export {
+    val name            = "export"
+    val Run: Permission = Permission(name, "run")
+
+    val list: List[Permission] = Run :: Nil
+  }
+
   val minimalPermissions: Set[Permission] =
     (Version.list ++
       Acls.list ++
@@ -172,6 +179,7 @@ object Permission {
       Views.list ++
       Storages.list ++
       Quotas.list ++
+      Export.list ++
       Supervision.list).toSet
 
   val adminPermissions: Set[Permission] =