Allow uploading mags, attachments to existing datasets

app/controllers/DatasetController.scala

@@ -118,7 +118,8 @@ case class ReserveAttachmentUploadToPathRequest(
     attachmentName: String,
     attachmentType: LayerAttachmentType.Value,
     attachmentDataformat: LayerAttachmentDataformat.Value,
-    pathPrefix: Option[UPath]
+    pathPrefix: Option[UPath],
+    overwritePending: Option[Boolean] = None
 )
 
 object ReserveAttachmentUploadToPathRequest {
@@ -166,14 +167,14 @@ class DatasetController @Inject()(userService: UserService,
                                   wKRemoteSegmentAnythingClient: WKRemoteSegmentAnythingClient,
                                   teamService: TeamService,
                                   datasetDAO: DatasetDAO,
-                                  datasetLayerAttachmentsDAO: DatasetLayerAttachmentsDAO,
+                                  datasetLayerAttachmentsDAO: DatasetLayerAttachmentDAO,
                                   datasetUploadToPathsService: UploadToPathsService,
                                   folderService: FolderService,
                                   thumbnailService: ThumbnailService,
                                   thumbnailCachingService: ThumbnailCachingService,
                                   usedStorageService: UsedStorageService,
                                   conf: WkConf,
-                                  datasetMagsDAO: DatasetMagsDAO,
+                                  datasetMagsDAO: DatasetMagDAO,
                                   slackNotificationService: SlackNotificationService,
                                   authenticationService: AccessibleBySwitchingService,
                                   analyticsService: AnalyticsService,
@@ -726,6 +727,7 @@ class DatasetController @Inject()(userService: UserService,
       for {
         dataset <- datasetDAO.findOne(datasetId) ?~> notFoundMessage(datasetId.toString) ~> NOT_FOUND
         _ <- Fox.assertTrue(datasetService.isEditableBy(dataset, Some(request.identity))) ?~> "notAllowed" ~> FORBIDDEN
+        _ <- datasetMagsDAO.findOneWithPendingUploadToPath(datasetId, request.body.layerName, request.body.mag) ?~> "dataset.finishMagUploadToPath.notPending"
         _ <- datasetMagsDAO.finishUploadToPath(datasetId, request.body.layerName, request.body.mag)
         dataStoreClient <- datasetService.clientFor(dataset)
         _ <- Fox.runIf(!dataset.isVirtual) {
@@ -753,10 +755,15 @@ class DatasetController @Inject()(userService: UserService,
       for {
         dataset <- datasetDAO.findOne(datasetId) ?~> notFoundMessage(datasetId.toString) ~> NOT_FOUND
         _ <- Fox.assertTrue(datasetService.isEditableBy(dataset, Some(request.identity))) ?~> "notAllowed" ~> FORBIDDEN
+        _ <- datasetLayerAttachmentsDAO.findOneWithPendingUploadToPath(
+          datasetId,
+          request.body.layerName,
+          request.body.attachmentType,
+          request.body.attachmentName) ?~> "dataset.finishAttachmentUploadToPath.notPending"
         _ <- datasetLayerAttachmentsDAO.finishUploadToPath(datasetId,
                                                            request.body.layerName,
-                                                           request.body.attachmentName,
-                                                           request.body.attachmentType)
+                                                           request.body.attachmentType,
+                                                           request.body.attachmentName)
         dataStoreClient <- datasetService.clientFor(dataset)
         _ <- Fox.runIf(!dataset.isVirtual) {
           for {

app/controllers/WKRemoteDataStoreController.scala

@@ -5,18 +5,26 @@ import com.scalableminds.util.objectid.ObjectId
 import com.scalableminds.util.time.Instant
 import com.scalableminds.util.tools.Fox
 import com.scalableminds.webknossos.datastore.controllers.JobExportProperties
+import com.scalableminds.webknossos.datastore.helpers.UPath
 import com.scalableminds.webknossos.datastore.models.UnfinishedUpload
 import com.scalableminds.webknossos.datastore.models.datasource.{
   DataSource,
   DataSourceId,
   DataSourceStatus,
+  LayerAttachmentType,
   UnusableDataSource
 }
 import com.scalableminds.webknossos.datastore.services.{DataSourcePathInfo, DataStoreStatus}
 import com.scalableminds.webknossos.datastore.services.uploading.{
+  AttachmentUploadAdditionalInfo,
+  AttachmentUploadInfo,
+  DatasetUploadAdditionalInfo,
+  DatasetUploadInfo,
+  MagUploadAdditionalInfo,
+  MagUploadInfo,
+  ReportAttachmentUploadParameters,
   ReportDatasetUploadParameters,
-  ReserveAdditionalInformation,
-  ReserveUploadInformation
+  ReportMagUploadParameters
 }
 import com.typesafe.scalalogging.LazyLogging
 import models.dataset._
@@ -46,6 +54,9 @@ class WKRemoteDataStoreController @Inject()(
     userDAO: UserDAO,
     teamDAO: TeamDAO,
     jobDAO: JobDAO,
+    datasetMagDAO: DatasetMagDAO,
+    datasetAttachmentDAO: DatasetLayerAttachmentDAO,
+    uploadToPathsService: UploadToPathsService,
     credentialDAO: CredentialDAO,
     wkSilhouetteEnvironment: WkSilhouetteEnvironment)(implicit ec: ExecutionContext, bodyParsers: PlayBodyParsers)
     extends Controller
@@ -54,45 +65,103 @@ class WKRemoteDataStoreController @Inject()(
   val bearerTokenService: WebknossosBearerTokenAuthenticatorService =
     wkSilhouetteEnvironment.combinedAuthenticatorService.tokenAuthenticatorService
 
-  def reserveDatasetUpload(name: String, key: String, token: String): Action[ReserveUploadInformation] =
-    Action.async(validateJson[ReserveUploadInformation]) { implicit request =>
+  def reserveDatasetUpload(name: String, key: String, token: String): Action[DatasetUploadInfo] =
+    Action.async(validateJson[DatasetUploadInfo]) { implicit request =>
       dataStoreService.validateAccess(name, key) { dataStore =>
         val uploadInfo = request.body
         for {
           user <- bearerTokenService.userForToken(token) ~> FORBIDDEN
-          organization <- organizationDAO.findOne(uploadInfo.organization)(GlobalAccessContext) ?~> Messages(
+          organization <- organizationDAO.findOne(uploadInfo.organizationId)(GlobalAccessContext) ?~> Messages(
             "organization.notFound",
-            uploadInfo.organization) ~> NOT_FOUND
+            uploadInfo.organizationId) ~> NOT_FOUND
           usedStorageBytes <- organizationDAO.getUsedStorage(organization._id)
           _ <- Fox.runOptional(organization.includedStorageBytes)(includedStorage =>
-            Fox.fromBool(usedStorageBytes + uploadInfo.totalFileSizeInBytes.getOrElse(0L) <= includedStorage)) ?~> "dataset.upload.storageExceeded" ~> FORBIDDEN
+            Fox.fromBool(usedStorageBytes + uploadInfo.resumableUploadInfo.totalFileSizeInBytes
+              .getOrElse(0L) <= includedStorage)) ?~> "dataset.upload.storageExceeded" ~> FORBIDDEN
           _ <- Fox.fromBool(organization._id == user._organization) ?~> "notAllowed" ~> FORBIDDEN
-          _ <- datasetService.assertValidDatasetName(uploadInfo.name)
+          _ <- datasetService.assertValidDatasetName(uploadInfo.datasetName)
           _ <- Fox.fromBool(dataStore.onlyAllowedOrganization.forall(_ == organization._id)) ?~> "dataset.upload.Datastore.restricted"
           _ <- 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))
+            datasetService.assertNewDatasetNameUnique(request.body.datasetName, organization._id))
           preliminaryDataSource = UnusableDataSource(DataSourceId("", ""), None, DataSourceStatus.notYetUploaded)
           dataset <- datasetService.createAndSetUpDataset(
-            uploadInfo.name,
+            uploadInfo.datasetName,
             dataStore,
             preliminaryDataSource,
             uploadInfo.folderId,
             user,
             isVirtual = uploadInfo.isVirtual.getOrElse(true),
             creationType = DatasetCreationType.Upload
           ) ?~> "dataset.upload.creation.failed"
-          _ <- datasetService.addInitialTeams(dataset, uploadInfo.initialTeams, user)(AuthorizedAccessContext(user))
-          additionalInfo = ReserveAdditionalInformation(dataset._id, dataset.directoryName)
+          _ <- datasetService.addInitialTeams(dataset, uploadInfo.initialTeamIds, user)(AuthorizedAccessContext(user))
+          additionalInfo = DatasetUploadAdditionalInfo(dataset._id, dataset.directoryName)
         } yield Ok(Json.toJson(additionalInfo))
       }
     }
 
-  def getUnfinishedUploadsForUser(name: String,
-                                  key: String,
-                                  token: String,
-                                  organizationId: String): Action[AnyContent] =
+  def reserveMagUpload(name: String, key: String, token: String): Action[MagUploadInfo] =
+    Action.async(validateJson[MagUploadInfo]) { implicit request =>
+      dataStoreService.validateAccess(name, key) { dataStore =>
+        // DS write access was asserted already at this point.
+        for {
+          user <- bearerTokenService.userForToken(token)
+          dataset <- datasetDAO.findOne(request.body.datasetId)(AuthorizedAccessContext(user))
+          _ <- Fox.fromBool(dataset.isVirtual) ?~> "dataset.reserveMagUpload.notVirtual"
+          (dataSource, dataLayer) <- datasetService.getDataSourceAndLayerFor(dataset, request.body.layerName)
+          _ <- Fox.fromBool(!dataLayer.mags.exists(_.mag.maxDim == request.body.mag.mag.maxDim)) ?~> s"New mag ${request.body.mag.mag} conflicts with existing mag of the layer."
+          _ <- Fox.fromBool(dataset._dataStore == dataStore.name) ?~> "Cannot upload mag to existing dataset via different datastore."
+          _ <- uploadToPathsService.handleExistingPendingMag(dataset,
+                                                             request.body.layerName,
+                                                             request.body.mag.mag,
+                                                             request.body.overwritePending)
+          _ <- datasetMagDAO.insertWithUploadPending(request.body.datasetId,
+                                                     request.body.layerName,
+                                                     request.body.mag.mag,
+                                                     request.body.mag.axisOrder,
+                                                     request.body.mag.channelIndex)
+        } yield Ok(Json.toJson(MagUploadAdditionalInfo(dataSource.id)))
+      }
+    }
+
+  def reserveAttachmentUpload(name: String, key: String, token: String): Action[AttachmentUploadInfo] =
+    Action.async(validateJson[AttachmentUploadInfo]) { implicit request =>
+      dataStoreService.validateAccess(name, key) { dataStore =>
+        // DS write access was asserted already at this point.
+        for {
+          user <- bearerTokenService.userForToken(token)
+          dataset <- datasetDAO.findOne(request.body.datasetId)(AuthorizedAccessContext(user))
+          _ <- Fox.fromBool(dataset.isVirtual) ?~> "dataset.reserveAttachmentUpload.notVirtual"
+          (dataSource, dataLayer) <- datasetService.getDataSourceAndLayerFor(dataset, request.body.layerName)
+          isSingletonAttachment = LayerAttachmentType.isSingletonAttachment(request.body.attachmentType)
+          existsError = if (isSingletonAttachment) "attachment.singleton.alreadyFilled" else "attachment.name.taken"
+          existingAttachmentOpt = dataLayer.attachments.flatMap(
+            _.getByTypeAndNameAlwaysReturnSingletons(request.body.attachmentType, request.body.attachment.name))
+          _ <- Fox.fromBool(existingAttachmentOpt.isEmpty) ?~> existsError
+          _ <- Fox.fromBool(dataset._dataStore == dataStore.name) ?~> "Cannot upload attachment to existing dataset via different datastore."
+          dummyAttachmentPath <- UPath.fromString("<pending upload>").toFox
+          _ <- uploadToPathsService.handleExistingPendingAttachment(dataset,
+                                                                    request.body.layerName,
+                                                                    request.body.attachmentType,
+                                                                    request.body.attachment.name,
+                                                                    request.body.overwritePending)
+          _ <- datasetAttachmentDAO.insertWithUploadPending(
+            request.body.datasetId,
+            request.body.layerName,
+            request.body.attachment.name,
+            request.body.attachmentType,
+            request.body.attachment.dataFormat,
+            dummyAttachmentPath
+          )
+        } yield Ok(Json.toJson(AttachmentUploadAdditionalInfo(dataSource.id)))
+      }
+    }
+
+  def getUnfinishedDatasetUploadsForUser(name: String,
+                                         key: String,
+                                         token: String,
+                                         organizationId: String): Action[AnyContent] =
     Action.async { implicit request =>
       dataStoreService.validateAccess(name, key) { _ =>
         for {
@@ -149,6 +218,48 @@ class WKRemoteDataStoreController @Inject()(
       }
     }
 
+  def reportMagUpload(name: String, key: String): Action[ReportMagUploadParameters] =
+    Action.async(validateJson[ReportMagUploadParameters]) { implicit request =>
+      dataStoreService.validateAccess(name, key) { _ =>
+        for {
+          dataset <- datasetDAO.findOne(request.body.datasetId)(GlobalAccessContext) ?~> Messages(
+            "dataset.notFound",
+            request.body.datasetId) ~> NOT_FOUND
+          _ <- datasetMagDAO.findOneWithPendingUpload(request.body.datasetId,
+                                                      request.body.layerName,
+                                                      request.body.mag.mag) ?~> "dataset.finishMagUpload.notPending"
+          _ <- request.body.mag.path.toFox ?~> "dataset.finishMagUpload.pathNotSet"
+          _ <- datasetMagDAO.finishUpload(request.body.datasetId, request.body.layerName, request.body.mag)
+          dataStoreClient <- datasetService.clientFor(dataset)(GlobalAccessContext)
+          _ <- dataStoreClient.invalidateDatasetInDSCache(dataset._id)
+          _ <- usedStorageService.refreshStorageReportForDataset(dataset)
+        } yield Ok
+      }
+    }
+
+  def reportAttachmentUpload(name: String, key: String): Action[ReportAttachmentUploadParameters] =
+    Action.async(validateJson[ReportAttachmentUploadParameters]) { implicit request =>
+      dataStoreService.validateAccess(name, key) { _ =>
+        for {
+          dataset <- datasetDAO.findOne(request.body.datasetId)(GlobalAccessContext) ?~> Messages(
+            "dataset.notFound",
+            request.body.datasetId) ~> NOT_FOUND
+          _ <- datasetAttachmentDAO.findOneWithPendingUpload(
+            request.body.datasetId,
+            request.body.layerName,
+            request.body.attachmentType,
+            request.body.attachment.name) ?~> "dataset.finishAttachmentUpload.notPending"
+          _ <- datasetAttachmentDAO.finishUpload(request.body.datasetId,
+                                                 request.body.layerName,
+                                                 request.body.attachmentType,
+                                                 request.body.attachment)
+          dataStoreClient <- datasetService.clientFor(dataset)(GlobalAccessContext)
+          _ <- dataStoreClient.invalidateDatasetInDSCache(dataset._id)
+          _ <- usedStorageService.refreshStorageReportForDataset(dataset)
+        } yield Ok
+      }
+    }
+
   def statusUpdate(name: String, key: String): Action[DataStoreStatus] = Action.async(validateJson[DataStoreStatus]) {
     implicit request =>
       dataStoreService.validateAccess(name, key) { _ =>