Skip to content

Commit

Permalink
Remove Tar archiving (#4286)
Browse files Browse the repository at this point in the history 
Remove Tar archiving
  • Loading branch information
@dantb
dantb authored Sep 21, 2023
1 parent d1492e9 commit dfd58f2
Show file tree
Hide file tree
Showing 18 changed files with 142 additions and 357 deletions.
91 changes: 38 additions & 53 deletions ...rchive/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/archive/ArchiveDownload.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ import monix.execution.Scheduler

import java.nio.ByteBuffer
import java.nio.charset.StandardCharsets
import akka.stream.alpakka.file.ArchiveMetadata

/**
* Archive download functionality.
Expand All @@ -55,10 +56,9 @@ trait ArchiveDownload {
* @param caller
* the caller to be used for checking for access
*/
def apply[M](
def apply(
value: ArchiveValue,
project: ProjectRef,
format: ArchiveFormat[M],
ignoreNotFound: Boolean
)(implicit caller: Caller, scheduler: Scheduler): IO[ArchiveRejection, AkkaSource]

Expand Down Expand Up @@ -96,18 +96,17 @@ object ArchiveDownload {
private val printer = Printer.spaces2.copy(dropNullValues = true)
private val sourcePrinter = Printer.spaces2.copy(dropNullValues = false)

override def apply[M](
override def apply(
value: ArchiveValue,
project: ProjectRef,
format: ArchiveFormat[M],
ignoreNotFound: Boolean
)(implicit caller: Caller, scheduler: Scheduler): IO[ArchiveRejection, AkkaSource] = {
for {
references <- value.resources.toList.traverse(toFullReference)
_ <- checkResourcePermissions(references, project)
contentStream <- resolveReferencesAsStream(references, project, ignoreNotFound, format)
contentStream <- resolveReferencesAsStream(references, project, ignoreNotFound)
} yield {
Source.fromGraph(StreamConverter(contentStream)).via(format.writeFlow)
Source.fromGraph(StreamConverter(contentStream)).via(Zip.writeFlow)
}
}

Expand All @@ -124,34 +123,29 @@ object ArchiveDownload {
}
}

private def resolveReferencesAsStream[M](
private def resolveReferencesAsStream(
references: List[FullArchiveReference],
project: ProjectRef,
ignoreNotFound: Boolean,
format: ArchiveFormat[M]
)(implicit caller: Caller): IO[ArchiveRejection, Stream[Task, (M, AkkaSource)]] = {
ignoreNotFound: Boolean
)(implicit caller: Caller): IO[ArchiveRejection, Stream[Task, (ArchiveMetadata, AkkaSource)]] = {
references
.traverseFilter {
case ref: FileReference => fileEntry(ref, project, format, ignoreNotFound)
case ref: ResourceReference => resourceEntry(ref, project, format, ignoreNotFound)
case ref: FileReference => fileEntry(ref, project, ignoreNotFound)
case ref: ResourceReference => resourceEntry(ref, project, ignoreNotFound)
}
.map(sortWith(format))
.map(sortWith)
.map(asStream)
}

private def sortWith[M](
format: ArchiveFormat[M]
)(list: List[(M, Task[AkkaSource])]): List[(M, Task[AkkaSource])] = {
list.sortBy { case (entry, _) =>
entry
}(format.ordering)
}
private def sortWith(list: List[(ArchiveMetadata, Task[AkkaSource])]): List[(ArchiveMetadata, Task[AkkaSource])] =
list.sortBy { case (entry, _) => entry }(Zip.ordering)

private def asStream[M](list: List[(M, Task[AkkaSource])]) = {
Stream.iterable(list).evalMap[Task, (M, AkkaSource)] { case (metadata, source) =>
private def asStream(
list: List[(ArchiveMetadata, Task[AkkaSource])]
): Stream[Task, (ArchiveMetadata, AkkaSource)] =
Stream.iterable(list).evalMap { case (metadata, source) =>
source.map(metadata -> _)
}
}

private def checkResourcePermissions(
refs: List[FullArchiveReference],
Expand All @@ -166,14 +160,13 @@ object ArchiveDownload {
)
.void

private def fileEntry[Metadata](
private def fileEntry(
ref: FileReference,
project: ProjectRef,
format: ArchiveFormat[Metadata],
ignoreNotFound: Boolean
)(implicit
caller: Caller
): IO[ArchiveRejection, Option[(Metadata, Task[AkkaSource])]] = {
): IO[ArchiveRejection, Option[(ArchiveMetadata, Task[AkkaSource])]] = {
val refProject = ref.project.getOrElse(project)
// the required permissions are checked for each file content fetch
val entry = fetchFileContent(ref.ref, refProject, caller)
Expand All @@ -184,21 +177,19 @@ object ArchiveDownload {
case FileRejection.AuthorizationFailed(addr, perm) => AuthorizationFailed(addr, perm)
case other => WrappedFileRejection(other)
}
.flatMap { case FileResponse(fileMetadata, content) =>
IO.fromEither(
pathOf(ref, project, format, fileMetadata.filename).map { path =>
val archiveMetadata = format.metadata(path, fileMetadata.bytes)
val contentTask: Task[AkkaSource] = content
.tapError(response =>
UIO.delay(
logger
.error(s"Error streaming file '${fileMetadata.filename}' for archive: ${response.value.value}")
)
)
.mapError(response => ArchiveDownloadError(fileMetadata.filename, response))
Some((archiveMetadata, contentTask))
}
)
.map { case FileResponse(fileMetadata, content) =>
val path = pathOf(ref, project, fileMetadata.filename)
val archiveMetadata = Zip.metadata(path)
val contentTask: Task[AkkaSource] = content
.tapError(response =>
UIO.delay(
logger
.error(s"Error streaming file '${fileMetadata.filename}' for archive: ${response.value.value}")
)
)
.mapError(response => ArchiveDownloadError(fileMetadata.filename, response))
Some((archiveMetadata, contentTask))

}
if (ignoreNotFound) entry.onErrorRecover { case _: ResourceNotFound => None }
else entry
Expand All @@ -207,27 +198,21 @@ object ArchiveDownload {
private def pathOf(
ref: FileReference,
project: ProjectRef,
format: ArchiveFormat[_],
filename: String
): Either[FilenameTooLong, String] =
ref.path.map { p => Right(p.value.toString) }.getOrElse {
): String =
ref.path.map(_.value.toString).getOrElse {
val p = ref.project.getOrElse(project)
Either.cond(
format != ArchiveFormat.Tar || filename.length < 100,
s"$p/file/$filename",
FilenameTooLong(ref.ref.original, p, filename)
)
s"$p/file/$filename"
}

private def resourceEntry[Metadata](
private def resourceEntry(
ref: ResourceReference,
project: ProjectRef,
format: ArchiveFormat[Metadata],
ignoreNotFound: Boolean
): IO[ArchiveRejection, Option[(Metadata, Task[AkkaSource])]] = {
): IO[ArchiveRejection, Option[(ArchiveMetadata, Task[AkkaSource])]] = {
val archiveEntry = resourceRefToByteString(ref, project).map { content =>
val path = pathOf(ref, project)
val metadata = format.metadata(path, content.length.toLong)
val metadata = Zip.metadata(path)
Some((metadata, Task.pure(Source.single(content))))
}
if (ignoreNotFound) archiveEntry.onErrorHandle { _: ResourceNotFound => None }
Expand Down
3 changes: 1 addition & 2 deletions ...ugins/archive/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/archive/Archives.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -166,13 +166,12 @@ class Archives(
def download(
id: IdSegment,
project: ProjectRef,
format: ArchiveFormat[_],
ignoreNotFound: Boolean
)(implicit caller: Caller, scheduler: Scheduler): IO[ArchiveRejection, AkkaSource] =
(for {
resource <- fetch(id, project)
value = resource.value
source <- archiveDownload(value.value, project, format, ignoreNotFound)
source <- archiveDownload(value.value, project, ignoreNotFound)
} yield source).span("downloadArchive")

private def eval(cmd: CreateArchive): IO[ArchiveRejection, ArchiveResource] =
Expand Down
94 changes: 0 additions & 94 deletions ...ve/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/archive/model/ArchiveFormat.scala
View file

This file was deleted.

6 changes: 0 additions & 6 deletions ...src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/archive/model/ArchiveRejection.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -69,11 +69,6 @@ object ArchiveRejection {
)).mkString("\n")
)

final case class FilenameTooLong(id: Iri, project: ProjectRef, fileName: String)
extends ArchiveRejection(
s"File '$id' in project '$project' has a file name '$fileName' exceeding the 100 character limit for a tar file."
)

/**
* Rejection returned when an archive doesn't exist.
*
Expand Down Expand Up @@ -201,7 +196,6 @@ object ArchiveRejection {
HttpResponseFields {
case ResourceAlreadyExists(_, _) => StatusCodes.Conflict
case InvalidResourceCollection(_, _, _) => StatusCodes.BadRequest
case FilenameTooLong(_, _, _) => StatusCodes.BadRequest
case ArchiveNotFound(_, _) => StatusCodes.NotFound
case InvalidArchiveId(_) => StatusCodes.BadRequest
case ProjectContextRejection(rejection) => rejection.status
Expand Down
29 changes: 29 additions & 0 deletions ...gins/archive/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/archive/model/Zip.scala
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,29 @@
package ch.epfl.bluebrain.nexus.delta.plugins.archive.model

import akka.NotUsed
import akka.http.scaladsl.model.{ContentType, HttpRequest, MediaTypes}
import akka.stream.alpakka.file.scaladsl.Archive
import akka.stream.alpakka.file.ArchiveMetadata
import akka.stream.scaladsl.{Flow, Source}
import akka.util.ByteString
import ch.epfl.bluebrain.nexus.delta.sdk.utils.HeadersUtils

/**
* Zip archive format
*
* @see
* https://en.wikipedia.org/wiki/ZIP_(file_format)#Limits for the limitations
*/
object Zip {
type WriteFlow[Metadata] = Flow[(Metadata, Source[ByteString, _]), ByteString, NotUsed]

lazy val contentType: ContentType = MediaTypes.`application/zip`

lazy val writeFlow: WriteFlow[ArchiveMetadata] = Archive.zip()

lazy val ordering: Ordering[ArchiveMetadata] = Ordering.by(md => md.filePath)

def metadata(filename: String): ArchiveMetadata = ArchiveMetadata.create(filename)

def checkHeader(req: HttpRequest): Boolean = HeadersUtils.matches(req.headers, Zip.contentType.mediaType)
}
Loading

0 comments on commit dfd58f2

Please sign in to comment.