Upload datasets to S3 (#8912)

### Description

- If configured, uploaded datasets are forwarded to S3 after unzipping
- layersToLink now reference the source dataset by datasetId (this
introduces api version 11)
- All newly uploaded datasets are virtual (meaning that the on-disk
datasource-properties.json is removed and the database entry is the
source of truth)
- Only exception: datasets with needsConversion=true. Will be cleaned up
in #8959
- layersToLink no longer creates symlinks but rather just entries in the
DB, just like it works for uploadToPaths
- Raw uploaded data is backed up inside of the .trash folder for
debugging purposes. It gets cleared regularly anyway.

### Steps to test:
- Configure S3 uploads using appropriate credentials in the
application.conf
- Upload a dataset
- View that dataset
- Also test that local filesystem case did not break
- Test uploading zipped dataset and non-zipped dataset
- Test tif upload with `yarn enable-jobs` and a worker
- test upload from libs (ideally with layersToLink)

### TODOs:
- [x] Handle uploading larger files with multipart upload
- [x] Fix uploaded files having "aws-chunked" encoding
- [x] Push updated application.conf without credentials
- [x] Test with more datasets
- [x] Do not upload files that are not referenced in layers (e.g.
datasource properties.json files)
- [x] use UPath for everything
- [x] is there still a case where uploaded datasets should not be
virtual? → no!
- [x] Make all uploaded datasets virtual
- [x] unify layers to link mechanism
- [x] api versioning (layersToLink adapter)
- [x] re-test with s3
- [x] delete local stuff or move to .trash?
- [x] make sure both old and new convert_to_wkw variants are supported

### Issues:
- fixes #8415 
- fixes #8934
- fixes #8893
- fixes #8749

- Follow-up issue #8979 

------
- [x] Added changelog entry (create a `$PR_NUMBER.md` file in
`unreleased_changes` or use `./tools/create-changelog-entry.py`)
- [x] Added migration guide entry if applicable (edit the same file as
for the changelog)
- [x] Removed dev-only changes
- [x] Considered [common edge
cases](../blob/master/.github/common_edge_cases.md)
- [x] Needs datastore update after deployment

---------

Co-authored-by: Norman Rzepka <[email protected]>
Co-authored-by: Florian M <[email protected]>
Co-authored-by: Florian M <[email protected]>
Co-authored-by: MichaelBuessemeyer <[email protected]>

Author: 4 people

app/controllers/DatasetController.scala

@@ -15,6 +15,7 @@ import com.scalableminds.webknossos.datastore.models.datasource.{
   LayerAttachmentType,
   UsableDataSource
 }
+import com.scalableminds.webknossos.datastore.services.uploading.LinkedLayerIdentifier
 import mail.{MailchimpClient, MailchimpTag}
 import models.analytics.{AnalyticsService, ChangeDatasetSettingsEvent, OpenDatasetEvent}
 import models.dataset._
@@ -25,6 +26,7 @@ import models.dataset.explore.{
 }
 import models.folder.FolderService
 import models.organization.OrganizationDAO
+import models.storage.UsedStorageService
 import models.team.{TeamDAO, TeamService}
 import models.user.{User, UserDAO, UserService}
 import play.api.i18n.{Messages, MessagesProvider}
@@ -54,12 +56,6 @@ object DatasetUpdateParameters extends TristateOptionJsonHelper {
     Json.configured(tristateOptionParsing).format[DatasetUpdateParameters]
 }
 
-case class LinkedLayerIdentifier(datasetId: ObjectId, layerName: String, newLayerName: Option[String] = None)
-
-object LinkedLayerIdentifier {
-  implicit val jsonFormat: OFormat[LinkedLayerIdentifier] = Json.format[LinkedLayerIdentifier]
-}
-
 case class ReserveDatasetUploadToPathsRequest(
     datasetName: String,
     layersToLink: Seq[LinkedLayerIdentifier],
@@ -121,7 +117,7 @@ object SegmentAnythingMaskParameters {
   implicit val jsonFormat: Format[SegmentAnythingMaskParameters] = Json.format[SegmentAnythingMaskParameters]
 }
 
-case class DataSourceRegistrationInfo(dataSource: UsableDataSource, folderId: Option[String], dataStoreName: String)
+case class DataSourceRegistrationInfo(dataSource: UsableDataSource, folderId: Option[ObjectId], dataStoreName: String)
 
 object DataSourceRegistrationInfo {
   implicit val jsonFormat: OFormat[DataSourceRegistrationInfo] = Json.format[DataSourceRegistrationInfo]
@@ -142,6 +138,7 @@ class DatasetController @Inject()(userService: UserService,
                                   folderService: FolderService,
                                   thumbnailService: ThumbnailService,
                                   thumbnailCachingService: ThumbnailCachingService,
+                                  usedStorageService: UsedStorageService,
                                   conf: WkConf,
                                   authenticationService: AccessibleBySwitchingService,
                                   analyticsService: AnalyticsService,
@@ -208,12 +205,13 @@ class DatasetController @Inject()(userService: UserService,
           folderService.getOrCreateFromPathLiteral(folderPath, request.identity._organization)) ?~> "dataset.explore.autoAdd.getFolder.failed"
         _ <- datasetService.assertValidDatasetName(request.body.datasetName)
         _ <- Fox.serialCombined(dataSource.dataLayers)(layer => datasetService.assertValidLayerNameLax(layer.name))
-        newDataset <- datasetService.createVirtualDataset(
+        newDataset <- datasetService.createAndSetUpDataset(
           request.body.datasetName,
           dataStore,
           dataSource,
-          folderIdOpt.map(_.toString),
-          request.identity
+          folderIdOpt,
+          request.identity,
+          isVirtual = true
         ) ?~> "dataset.explore.autoAdd.failed"
       } yield Ok(Json.toJson(newDataset._id))
     }
@@ -229,13 +227,19 @@ class DatasetController @Inject()(userService: UserService,
         _ <- Fox.fromBool(isTeamManagerOrAdmin || user.isDatasetManager) ~> FORBIDDEN
         _ <- Fox.fromBool(request.body.dataSource.dataLayers.nonEmpty) ?~> "dataset.explore.zeroLayers"
         _ <- datasetService.validatePaths(request.body.dataSource.allExplicitPaths, dataStore) ?~> "dataSource.add.pathsNotAllowed"
-        dataset <- datasetService.createVirtualDataset(
+        dataset <- datasetService.createAndSetUpDataset(
           name,
           dataStore,
           request.body.dataSource,
           request.body.folderId,
-          user
+          user,
+          isVirtual = true
         )
+        _ = datasetService.trackNewDataset(dataset,
+                                           user,
+                                           needsConversion = false,
+                                           datasetSizeBytes = 0,
+                                           viaAddRoute = false)
       } yield Ok(Json.obj("newDatasetId" -> dataset._id))
     }
 
@@ -654,6 +658,12 @@ class DatasetController @Inject()(userService: UserService,
           dataset.status == DataSourceStatus.notYetUploadedToPaths || dataset.status == DataSourceStatus.notYetUploaded) ?~> s"Dataset is not in uploading-to-paths status, got ${dataset.status}."
         _ <- Fox.fromBool(!dataset.isUsable) ?~> s"Dataset is already marked as usable."
         _ <- datasetDAO.updateDatasetStatusByDatasetId(datasetId, newStatus = "", isUsable = true)
+        _ <- usedStorageService.refreshStorageReportForDataset(dataset)
+        _ = datasetService.trackNewDataset(dataset,
+                                           request.identity,
+                                           needsConversion = false,
+                                           datasetSizeBytes = 0,
+                                           viaAddRoute = false)
       } yield Ok
     }
 

app/controllers/WKRemoteDataStoreController.scala

@@ -1,37 +1,37 @@
 package controllers
 
-import com.scalableminds.util.accesscontext.{AuthorizedAccessContext, GlobalAccessContext}
+import com.scalableminds.util.accesscontext.{AuthorizedAccessContext, DBAccessContext, GlobalAccessContext}
 import com.scalableminds.util.objectid.ObjectId
 import com.scalableminds.util.time.Instant
 import com.scalableminds.util.tools.{Fox, Full}
 import com.scalableminds.webknossos.datastore.controllers.JobExportProperties
 import com.scalableminds.webknossos.datastore.helpers.{LayerMagLinkInfo, MagLinkInfo}
 import com.scalableminds.webknossos.datastore.models.UnfinishedUpload
-import com.scalableminds.webknossos.datastore.models.datasource.{DataSourceId, DataSource}
+import com.scalableminds.webknossos.datastore.models.datasource.{
+  DataSource,
+  DataSourceId,
+  DataSourceStatus,
+  UnusableDataSource
+}
 import com.scalableminds.webknossos.datastore.services.{DataSourcePathInfo, DataStoreStatus}
 import com.scalableminds.webknossos.datastore.services.uploading.{
-  LegacyLinkedLayerIdentifier,
+  ReportDatasetUploadParameters,
   ReserveAdditionalInformation,
   ReserveUploadInformation
 }
 import com.typesafe.scalalogging.LazyLogging
-import mail.{MailchimpClient, MailchimpTag}
-import models.analytics.{AnalyticsService, UploadDatasetEvent}
 import models.annotation.AnnotationDAO
 import models.dataset._
 import models.dataset.credential.CredentialDAO
-import models.folder.FolderDAO
 import models.job.JobDAO
 import models.organization.OrganizationDAO
 import models.storage.UsedStorageService
 import models.team.TeamDAO
-import models.user.{MultiUserDAO, User, UserDAO, UserService}
-import play.api.i18n.{Messages, MessagesProvider}
+import models.user.UserDAO
+import play.api.i18n.Messages
 import play.api.libs.json.Json
 import play.api.mvc.{Action, AnyContent, PlayBodyParsers}
 import security.{WebknossosBearerTokenAuthenticatorService, WkSilhouetteEnvironment}
-import telemetry.SlackNotificationService
-import utils.WkConf
 
 import scala.concurrent.duration.DurationInt
 import javax.inject.Inject
@@ -41,22 +41,16 @@ class WKRemoteDataStoreController @Inject()(
     datasetService: DatasetService,
     dataStoreService: DataStoreService,
     dataStoreDAO: DataStoreDAO,
-    analyticsService: AnalyticsService,
-    userService: UserService,
     organizationDAO: OrganizationDAO,
     usedStorageService: UsedStorageService,
+    layerToLinkService: LayerToLinkService,
     datasetDAO: DatasetDAO,
     datasetLayerDAO: DatasetLayerDAO,
     userDAO: UserDAO,
-    folderDAO: FolderDAO,
     teamDAO: TeamDAO,
     jobDAO: JobDAO,
-    multiUserDAO: MultiUserDAO,
     credentialDAO: CredentialDAO,
     annotationDAO: AnnotationDAO,
-    mailchimpClient: MailchimpClient,
-    slackNotificationService: SlackNotificationService,
-    conf: WkConf,
     wkSilhouetteEnvironment: WkSilhouetteEnvironment)(implicit ec: ExecutionContext, bodyParsers: PlayBodyParsers)
     extends Controller
     with LazyLogging {
@@ -79,26 +73,22 @@ class WKRemoteDataStoreController @Inject()(
           _ <- Fox.fromBool(organization._id == user._organization) ?~> "notAllowed" ~> FORBIDDEN
           _ <- datasetService.assertValidDatasetName(uploadInfo.name)
           _ <- Fox.fromBool(dataStore.onlyAllowedOrganization.forall(_ == organization._id)) ?~> "dataset.upload.Datastore.restricted"
-          folderId = uploadInfo.folderId.getOrElse(organization._rootFolder)
-          _ <- folderDAO.assertUpdateAccess(folderId)(AuthorizedAccessContext(user)) ?~> "folder.noWriteAccess"
-          layersToLinkWithDirectoryName <- Fox.serialCombined(uploadInfo.layersToLink.getOrElse(List.empty))(l =>
-            validateLayerToLink(l, user)) ?~> "dataset.upload.invalidLinkedLayers"
-          newDatasetId = ObjectId.generate
+          _ <- Fox.serialCombined(uploadInfo.layersToLink.getOrElse(List.empty))(l =>
+            layerToLinkService.validateLayerToLink(l, user)) ?~> "dataset.upload.invalidLinkedLayers"
           _ <- Fox.runIf(request.body.requireUniqueName.getOrElse(false))(
             datasetService.assertNewDatasetNameUnique(request.body.name, organization._id))
-          dataset <- datasetService.createPreliminaryDataset(newDatasetId,
-                                                             uploadInfo.name,
-                                                             datasetService.generateDirectoryName(uploadInfo.name,
-                                                                                                  newDatasetId),
-                                                             uploadInfo.organization,
-                                                             dataStore) ?~> "dataset.upload.creation.failed"
-          _ <- datasetDAO.updateFolder(dataset._id, folderId)(GlobalAccessContext)
+          preliminaryDataSource = UnusableDataSource(DataSourceId("", ""), None, DataSourceStatus.notYetUploaded)
+          dataset <- datasetService.createAndSetUpDataset(
+            uploadInfo.name,
+            dataStore,
+            preliminaryDataSource,
+            uploadInfo.folderId,
+            user,
+            // For the moment, the convert_to_wkw job can only fill the dataset if it is not virtual.
+            isVirtual = !uploadInfo.needsConversion.getOrElse(false)
+          ) ?~> "dataset.upload.creation.failed"
           _ <- datasetService.addInitialTeams(dataset, uploadInfo.initialTeams, user)(AuthorizedAccessContext(user))
-          _ <- datasetService.addUploader(dataset, user._id)(AuthorizedAccessContext(user))
-          additionalInfo = ReserveAdditionalInformation(dataset._id,
-                                                        dataset.directoryName,
-                                                        if (layersToLinkWithDirectoryName.isEmpty) None
-                                                        else Some(layersToLinkWithDirectoryName))
+          additionalInfo = ReserveAdditionalInformation(dataset._id, dataset.directoryName)
         } yield Ok(Json.toJson(additionalInfo))
       }
     }
@@ -132,62 +122,37 @@ class WKRemoteDataStoreController @Inject()(
       }
     }
 
-  private def validateLayerToLink(layerIdentifier: LegacyLinkedLayerIdentifier, requestingUser: User)(
-      implicit ec: ExecutionContext,
-      m: MessagesProvider): Fox[LegacyLinkedLayerIdentifier] =
-    for {
-      organization <- organizationDAO.findOne(layerIdentifier.getOrganizationId)(GlobalAccessContext) ?~> Messages(
-        "organization.notFound",
-        layerIdentifier.getOrganizationId) ~> NOT_FOUND
-      datasetBox <- datasetDAO
-        .findOneByNameAndOrganization(layerIdentifier.dataSetName, organization._id)(
-          AuthorizedAccessContext(requestingUser))
-        .shiftBox
-      dataset <- datasetBox match {
-        case Full(ds) => Fox.successful(ds)
-        case _ =>
-          ObjectId
-            .fromString(layerIdentifier.dataSetName)
-            .flatMap(interpretedAsId => datasetDAO.findOne(interpretedAsId)(AuthorizedAccessContext(requestingUser))) ?~> Messages(
-            "dataset.notFound",
-            layerIdentifier.dataSetName)
-      }
-      isTeamManagerOrAdmin <- userService.isTeamManagerOrAdminOfOrg(requestingUser, dataset._organization)
-      _ <- Fox.fromBool(isTeamManagerOrAdmin || requestingUser.isDatasetManager || dataset.isPublic) ?~> "dataset.upload.linkRestricted"
-    } yield layerIdentifier.copy(datasetDirectoryName = Some(dataset.directoryName))
-
   def reportDatasetUpload(name: String,
                           key: String,
                           token: String,
-                          datasetDirectoryName: String,
-                          datasetSizeBytes: Long,
-                          needsConversion: Boolean,
-                          viaAddRoute: Boolean): Action[AnyContent] =
-    Action.async { implicit request =>
-      dataStoreService.validateAccess(name, key) { dataStore =>
+                          datasetId: ObjectId): Action[ReportDatasetUploadParameters] =
+    Action.async(validateJson[ReportDatasetUploadParameters]) { implicit request =>
+      dataStoreService.validateAccess(name, key) { _ =>
         for {
           user <- bearerTokenService.userForToken(token)
-          dataset <- datasetDAO.findOneByDirectoryNameAndOrganization(datasetDirectoryName, user._organization)(
-            GlobalAccessContext) ?~> Messages("dataset.notFound", datasetDirectoryName) ~> NOT_FOUND
-          _ <- Fox.runIf(!needsConversion && !viaAddRoute)(usedStorageService.refreshStorageReportForDataset(dataset))
-          _ <- Fox.runIf(!needsConversion)(logUploadToSlack(user, dataset._id, viaAddRoute))
-          _ = analyticsService.track(UploadDatasetEvent(user, dataset, dataStore, datasetSizeBytes))
-          _ = if (!needsConversion) mailchimpClient.tagUser(user, MailchimpTag.HasUploadedOwnDataset)
-        } yield Ok(Json.obj("id" -> dataset._id))
+          dataset <- datasetDAO.findOne(datasetId)(GlobalAccessContext) ?~> Messages("dataset.notFound", datasetId) ~> NOT_FOUND
+          _ <- Fox.runIf(!request.body.needsConversion)(usedStorageService.refreshStorageReportForDataset(dataset))
+          _ = datasetService.trackNewDataset(dataset,
+                                             user,
+                                             request.body.needsConversion,
+                                             request.body.datasetSizeBytes,
+                                             viaAddRoute = false)
+          dataSourceWithLinkedLayersOpt <- Fox.runOptional(request.body.dataSourceOpt) {
+            implicit val ctx: DBAccessContext = AuthorizedAccessContext(user)
+            layerToLinkService.addLayersToLinkToDataSource(_, request.body.layersToLink)
+          }
+          _ <- Fox.runOptional(dataSourceWithLinkedLayersOpt) { dataSource =>
+            logger.info(s"Updating dataset $datasetId in database after upload reported from datastore $name.")
+            datasetDAO.updateDataSource(datasetId,
+                                        dataset._dataStore,
+                                        dataSource.hashCode(),
+                                        dataSource,
+                                        isUsable = true)(GlobalAccessContext)
+          }
+        } yield Ok
       }
     }
 
-  private def logUploadToSlack(user: User, datasetId: ObjectId, viaAddRoute: Boolean): Fox[Unit] =
-    for {
-      organization <- organizationDAO.findOne(user._organization)(GlobalAccessContext)
-      multiUser <- multiUserDAO.findOne(user._multiUser)(GlobalAccessContext)
-      resultLink = s"${conf.Http.uri}/datasets/$datasetId"
-      addLabel = if (viaAddRoute) "(via explore+add)" else "(upload without conversion)"
-      superUserLabel = if (multiUser.isSuperUser) " (for superuser)" else ""
-      _ = slackNotificationService.info(s"Dataset added $addLabel$superUserLabel",
-                                        s"For organization: ${organization.name}. <$resultLink|Result>")
-    } yield ()
-
   def statusUpdate(name: String, key: String): Action[DataStoreStatus] = Action.async(validateJson[DataStoreStatus]) {
     implicit request =>
       dataStoreService.validateAccess(name, key) { _ =>
@@ -242,11 +207,11 @@ class WKRemoteDataStoreController @Inject()(
   /**
     * Called by the datastore after a dataset has been deleted on disk.
     */
-  def deleteDataset(name: String, key: String): Action[DataSourceId] = Action.async(validateJson[DataSourceId]) {
+  def deleteDataset(name: String, key: String): Action[ObjectId] = Action.async(validateJson[ObjectId]) {
     implicit request =>
       dataStoreService.validateAccess(name, key) { _ =>
         for {
-          existingDatasetBox <- datasetDAO.findOneByDataSourceId(request.body)(GlobalAccessContext).shiftBox
+          existingDatasetBox <- datasetDAO.findOne(request.body)(GlobalAccessContext).shiftBox
           _ <- existingDatasetBox match {
             case Full(dataset) =>
               for {
@@ -272,7 +237,7 @@ class WKRemoteDataStoreController @Inject()(
             "organization.notFound",
             organizationId) ~> NOT_FOUND
           dataset <- datasetDAO.findOneByNameAndOrganization(datasetDirectoryName, organization._id)(
-            GlobalAccessContext)
+            GlobalAccessContext) ?~> Messages("dataset.notFound", datasetDirectoryName)
         } yield Ok(Json.toJson(dataset._id))
       }
     }

app/models/dataset/ComposeService.scala

@@ -47,11 +47,12 @@ class ComposeService @Inject()(datasetDAO: DatasetDAO, dataStoreDAO: DataStoreDA
       _ <- Fox.assertTrue(isComposable(composeRequest)) ?~> "Datasets are not composable, they are not on the same data store"
       dataSource <- createDatasource(composeRequest, composeRequest.newDatasetName, composeRequest.organizationId)
       dataStore <- dataStoreDAO.findOneWithUploadsAllowed
-      dataset <- datasetService.createVirtualDataset(composeRequest.newDatasetName,
-                                                     dataStore,
-                                                     dataSource,
-                                                     Some(composeRequest.targetFolderId.toString),
-                                                     user)
+      dataset <- datasetService.createAndSetUpDataset(composeRequest.newDatasetName,
+                                                      dataStore,
+                                                      dataSource,
+                                                      Some(composeRequest.targetFolderId),
+                                                      user,
+                                                      isVirtual = true)
 
     } yield (dataSource, dataset._id)
 
@@ -67,13 +68,9 @@ class ComposeService @Inject()(datasetDAO: DatasetDAO, dataStoreDAO: DataStoreDA
           case Some(c) => Some(c ++ composeLayer.transformations.toList)
           case None    => Some(composeLayer.transformations.toList)
       }
-      editedLayer: StaticLayer <- layer match {
-        case l: StaticLayer =>
-          Fox.successful(
-            l.mapped(name = composeLayer.newName,
-                     coordinateTransformations = applyCoordinateTransformations(l.coordinateTransformations)))
-        case _ => Fox.failure("Unsupported layer type for composition: " + layer.getClass.getSimpleName)
-      }
+      editedLayer = layer.mapped(name = composeLayer.newName,
+                                 coordinateTransformations =
+                                   applyCoordinateTransformations(layer.coordinateTransformations))
     } yield (editedLayer, usableDataSource.scale)
 
   private def isComposable(composeRequest: ComposeRequest)(implicit ctx: DBAccessContext): Fox[Boolean] =