Explore remote datasets as virtual datasets

Author: frcroth

app/controllers/DatasetController.scala

@@ -145,10 +145,10 @@ class DatasetController @Inject()(userService: UserService,
         _ <- Fox.fromBool(dataSource.dataLayers.nonEmpty) ?~> "dataset.explore.zeroLayers"
         folderIdOpt <- Fox.runOptional(request.body.folderPath)(folderPath =>
           folderService.getOrCreateFromPathLiteral(folderPath, request.identity._organization)) ?~> "dataset.explore.autoAdd.getFolder.failed"
-        _ <- wkExploreRemoteLayerService.addRemoteDatasource(dataSource,
-                                                             request.body.datasetName,
-                                                             request.identity,
-                                                             folderIdOpt) ?~> "dataset.explore.autoAdd.failed"
+        _ <- wkExploreRemoteLayerService.addRemoteDatasourceToDatabase(dataSource,
+                                                                       request.body.datasetName,
+                                                                       request.identity,
+                                                                       folderIdOpt) ?~> "dataset.explore.autoAdd.failed"
       } yield Ok
     }
 

app/controllers/WKRemoteDataStoreController.scala

@@ -9,7 +9,7 @@ import com.scalableminds.webknossos.datastore.helpers.{LayerMagLinkInfo, MagLink
 import com.scalableminds.webknossos.datastore.models.UnfinishedUpload
 import com.scalableminds.webknossos.datastore.models.datasource.DataSourceId
 import com.scalableminds.webknossos.datastore.models.datasource.inbox.{InboxDataSourceLike => InboxDataSource}
-import com.scalableminds.webknossos.datastore.services.{DataSourcePathInfo, DataStoreStatus}
+import com.scalableminds.webknossos.datastore.services.{DataSourcePathInfo, DataSourceRegistrationInfo, DataStoreStatus}
 import com.scalableminds.webknossos.datastore.services.uploading.{
   LinkedLayerIdentifier,
   ReserveAdditionalInformation,
@@ -259,7 +259,7 @@ class WKRemoteDataStoreController @Inject()(
       }
     }
 
-  def getDatasource(name: String, key: String, datasetId: ObjectId): Action[AnyContent] =
+  def getDataSource(name: String, key: String, datasetId: ObjectId): Action[AnyContent] =
     Action.async { implicit request =>
       dataStoreService.validateAccess(name, key) { _ =>
         for {
@@ -270,6 +270,34 @@ class WKRemoteDataStoreController @Inject()(
 
     }
 
+  // Register a datasource from the datastore as a dataset in the database.
+  // This is called when adding remote virtual datasets (that should only exist in the database)
+  // by the data store after exploration.
+  def registerDataSource(name: String,
+                         key: String,
+                         organizationId: String,
+                         directoryName: String,
+                         token: String): Action[DataSourceRegistrationInfo] =
+    Action.async(validateJson[DataSourceRegistrationInfo]) { implicit request =>
+      dataStoreService.validateAccess(name, key) { dataStore =>
+        for {
+          user <- bearerTokenService.userForToken(token)
+          organization <- organizationDAO.findOne(organizationId)(GlobalAccessContext) ?~> Messages(
+            "organization.notFound",
+            organizationId) ~> NOT_FOUND
+          _ <- Fox.fromBool(organization._id == user._organization) ?~> "notAllowed" ~> FORBIDDEN
+          dataset <- datasetService.createVirtualDataset(
+            directoryName,
+            organizationId,
+            dataStore,
+            request.body.dataSource,
+            request.body.folderId,
+            user
+          )
+        } yield Ok(dataset._id.toString)
+      }
+    }
+
   def jobExportProperties(name: String, key: String, jobId: ObjectId): Action[AnyContent] = Action.async {
     implicit request =>
       dataStoreService.validateAccess(name, key) { _ =>

app/models/dataset/DataStore.scala

@@ -79,8 +79,7 @@ class DataStoreService @Inject()(dataStoreDAO: DataStoreDAO, jobService: JobServ
 
   def validateAccess(name: String, key: String)(block: DataStore => Future[Result])(
       implicit m: MessagesProvider): Fox[Result] =
-    Fox.fromFuture((for {
-      dataStore <- dataStoreDAO.findOneByName(name)(GlobalAccessContext)
+    Fox.fromFuture((for {dataStore <- dataStoreDAO.findOneByName(name)(GlobalAccessContext)
       _ <- Fox.fromBool(key == dataStore.key)
       result <- Fox.fromFuture(block(dataStore))
     } yield result).getOrElse(Forbidden(Json.obj("granted" -> false, "msg" -> Messages("dataStore.notFound")))))

app/models/dataset/DatasetService.scala

@@ -1,6 +1,6 @@
 package models.dataset
 
-import com.scalableminds.util.accesscontext.{DBAccessContext, 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, FoxImplicits}
@@ -23,6 +23,7 @@ import com.scalableminds.webknossos.datastore.models.datasource.{
   AbstractDataLayer,
   AbstractSegmentationLayer,
   DataFormat,
+  DataSource,
   DataSourceId,
   GenericDataSource,
   DataLayerLike => DataLayer
@@ -36,6 +37,7 @@ import models.team._
 import models.user.{User, UserService}
 import net.liftweb.common.Box.tryo
 import net.liftweb.common.{Empty, EmptyBox, Full}
+import play.api.i18n.Messages
 import play.api.libs.json.{JsObject, Json}
 import security.RandomIDGenerator
 import utils.WkConf
@@ -97,6 +99,34 @@ class DatasetService @Inject()(organizationDAO: OrganizationDAO,
     } yield newDataset
   }
 
+  private def virtualRemoteDatasetStatus = "Virtual remote dataset"
+
+  def createVirtualDataset(datasetName: String,
+                           organizationId: String,
+                           dataStore: DataStore,
+                           dataSource: DataSource,
+                           folderId: Option[String],
+                           user: User): Fox[Dataset] =
+    for {
+      _ <- assertValidDatasetName(datasetName)
+      isDatasetNameAlreadyTaken <- datasetDAO.doesDatasetDirectoryExistInOrganization(datasetName, organizationId)(
+        GlobalAccessContext)
+      _ <- Fox.fromBool(!isDatasetNameAlreadyTaken) ?~> "dataset.name.alreadyTaken"
+      organization <- organizationDAO.findOne(organizationId)(GlobalAccessContext) ?~> "organization.notFound"
+      folderId <- ObjectId.fromString(folderId.getOrElse(organization._rootFolder.toString)) ?~> "dataset.upload.folderId.invalid"
+      _ <- folderDAO.assertUpdateAccess(folderId)(AuthorizedAccessContext(user)) ?~> "folder.noWriteAccess"
+      newDatasetId = ObjectId.generate
+      abstractDataSource = dataSource.copy(dataLayers = dataSource.dataLayers.map(_.asAbstractLayer))
+      dataset <- createDataset(dataStore,
+                               newDatasetId,
+                               datasetName,
+                               abstractDataSource,
+                               status = Some(virtualRemoteDatasetStatus))
+      datasetId = dataset._id
+      _ <- datasetDAO.updateFolder(datasetId, folderId)(GlobalAccessContext)
+      _ <- addUploader(dataset, user._id)(GlobalAccessContext)
+    } yield dataset
+
   def getAllUnfinishedDatasetUploadsOfUser(userId: ObjectId, organizationId: String)(
       implicit ctx: DBAccessContext): Fox[List[DatasetCompactInfo]] =
     datasetDAO.findAllCompactWithSearch(
@@ -114,7 +144,8 @@ class DatasetService @Inject()(organizationDAO: OrganizationDAO,
       datasetId: ObjectId,
       datasetName: String,
       dataSource: InboxDataSource,
-      publication: Option[ObjectId] = None
+      publication: Option[ObjectId] = None,
+      status: Option[String] = None
   ): Fox[Dataset] = {
     implicit val ctx: DBAccessContext = GlobalAccessContext
     val metadata =
@@ -147,7 +178,7 @@ class DatasetService @Inject()(organizationDAO: OrganizationDAO,
         name = datasetName,
         voxelSize = dataSource.voxelSizeOpt,
         sharingToken = None,
-        status = dataSource.statusOpt.getOrElse(""),
+        status = status.orElse(dataSource.statusOpt).getOrElse(""),
         logoUrl = None,
         metadata = metadata
       )

app/models/dataset/WKRemoteDataStoreClient.scala

@@ -82,18 +82,6 @@ class WKRemoteDataStoreClient(dataStore: DataStore, rpc: RPC) extends LazyLoggin
       .silent
       .getWithJsonResponse[List[DirectoryStorageReport]]
 
-  def addDataSource(organizationId: String,
-                    datasetName: String,
-                    dataSource: GenericDataSource[DataLayer],
-                    folderId: Option[ObjectId],
-                    userToken: String): Fox[Unit] =
-    for {
-      _ <- rpc(s"${dataStore.url}/data/datasets/$organizationId/$datasetName")
-        .addQueryString("token" -> userToken)
-        .addQueryStringOptional("folderId", folderId.map(_.toString))
-        .postJson(dataSource)
-    } yield ()
-
   def hasSegmentIndexFile(organizationId: String, datasetName: String, layerName: String)(
       implicit ec: ExecutionContext): Fox[Boolean] = {
     val cacheKey = (organizationId, datasetName, layerName)

app/models/dataset/explore/WKExploreRemoteLayerService.scala

@@ -105,17 +105,22 @@ class WKExploreRemoteLayerService @Inject()(credentialService: CredentialService
       credentialId <- Fox.runOptional(credentialOpt)(c => credentialService.insertOne(c)) ?~> "dataVault.credential.insert.failed"
     } yield credentialId
 
-  def addRemoteDatasource(dataSource: GenericDataSource[DataLayer],
-                          datasetName: String,
-                          user: User,
-                          folderId: Option[ObjectId])(implicit ctx: DBAccessContext): Fox[Unit] =
+  def addRemoteDatasourceToDatabase(dataSource: GenericDataSource[DataLayer],
+                                    datasetName: String,
+                                    user: User,
+                                    folderId: Option[ObjectId])(implicit ctx: DBAccessContext): Fox[Unit] =
     for {
-      organization <- organizationDAO.findOne(user._organization)
       dataStore <- dataStoreDAO.findOneWithUploadsAllowed
+      organizationId = user._organization
       _ <- datasetService.assertValidDatasetName(datasetName)
-      client = new WKRemoteDataStoreClient(dataStore, rpc)
-      userToken <- bearerTokenService.createAndInitDataStoreTokenForUser(user)
-      _ <- client.addDataSource(organization._id, datasetName, dataSource, folderId, userToken)
-    } yield ()
+      datasetId <- datasetService.createVirtualDataset(
+        dataSource.id.directoryName,
+        organizationId,
+        dataStore,
+        dataSource,
+        folderId.map(_.toString),
+        user
+      )
+    } yield datasetId
 
 }

conf/webknossos.latest.routes

@@ -110,7 +110,8 @@ PUT           /datastores/:name/datasource
 PUT           /datastores/:name/datasources                                                     controllers.WKRemoteDataStoreController.updateAll(name: String, key: String)
 PUT           /datastores/:name/datasources/paths                                               controllers.WKRemoteDataStoreController.updatePaths(name: String, key: String)
 GET           /datastores/:name/datasources/:organizationId/:directoryName/paths                controllers.WKRemoteDataStoreController.getPaths(name: String, key: String, organizationId: String, directoryName: String)
-GET           /datastores/:name/datasources/:datasetId                                          controllers.WKRemoteDataStoreController.getDatasource(name: String, key: String, datasetId: ObjectId)
+GET           /datastores/:name/datasources/:datasetId                                          controllers.WKRemoteDataStoreController.getDataSource(name: String, key: String, datasetId: ObjectId)
+POST          /datastores/:name/datasources/:organizationId/:directoryName                      controllers.WKRemoteDataStoreController.registerDataSource(name: String, key: String, organizationId: String, directoryName: String, token: String)
 PATCH         /datastores/:name/status                                                          controllers.WKRemoteDataStoreController.statusUpdate(name: String, key: String)
 POST          /datastores/:name/reserveUpload                                                   controllers.WKRemoteDataStoreController.reserveDatasetUpload(name: String, key: String, token: String)
 GET           /datastores/:name/getUnfinishedUploadsForUser                                     controllers.WKRemoteDataStoreController.getUnfinishedUploadsForUser(name: String, key: String, token: String, organizationName: String)

webknossos-datastore/app/com/scalableminds/webknossos/datastore/controllers/DataSourceController.scala

@@ -410,32 +410,17 @@ class DataSourceController @Inject()(
       }
     }
 
-  // Stores a remote dataset in the database.
+  // Called by the frontend after the user has set datasetName / FolderId of an explored dataSource
+  // Add this data source to the WK database
   def add(organizationId: String, datasetName: String, folderId: Option[String]): Action[DataSource] =
     Action.async(validateJson[DataSource]) { implicit request =>
       accessTokenService.validateAccessFromTokenContext(UserAccessRequest.administrateDataSources) {
         for {
-          reservedAdditionalInfo <- dsRemoteWebknossosClient.reserveDataSourceUpload(
-            ReserveUploadInformation(
-              uploadId = "", // Set by core backend
-              name = datasetName,
-              organization = organizationId,
-              totalFileCount = 1,
-              filePaths = None,
-              totalFileSizeInBytes = None,
-              layersToLink = None,
-              initialTeams = List.empty,
-              folderId = folderId,
-              requireUniqueName = Some(false),
-            )
-          ) ?~> "dataset.upload.validation.failed"
-          datasourceId = DataSourceId(reservedAdditionalInfo.directoryName, organizationId)
-          _ <- dataSourceService.updateDataSource(request.body.copy(id = datasourceId), expectExisting = false)
-          uploadedDatasetId <- dsRemoteWebknossosClient.reportUpload(datasourceId,
-                                                                     0L,
-                                                                     needsConversion = false,
-                                                                     viaAddRoute = true) ?~> "reportUpload.failed"
-        } yield Ok(Json.obj("newDatasetId" -> uploadedDatasetId))
+          _ <- Fox.successful(())
+          dataSourceId = DataSourceId(datasetName, organizationId)
+          dataSource = request.body.copy(id = dataSourceId)
+          datasetId <- dsRemoteWebknossosClient.registerDataSource(dataSource, dataSourceId, folderId) ?~> "dataset.add.failed"
+        } yield Ok(Json.obj("newDatasetId" -> datasetId))
       }
     }
 

webknossos-datastore/app/com/scalableminds/webknossos/datastore/dataformats/layers/N5DataLayers.scala

@@ -37,7 +37,24 @@ case class N5DataLayer(
     override val numChannels: Option[Int] = Some(1),
     additionalAxes: Option[Seq[AdditionalAxis]] = None,
     attachments: Option[DatasetLayerAttachments] = None,
-) extends N5Layer
+) extends N5Layer {
+  override def asAbstractLayer: DataLayerLike =
+    AbstractDataLayer(
+      name,
+      category,
+      boundingBox,
+      resolutions,
+      elementClass,
+      defaultViewConfiguration,
+      adminViewConfiguration,
+      coordinateTransformations,
+      additionalAxes,
+      attachments,
+      Some(mags),
+      numChannels,
+      Some(dataFormat)
+    )
+}
 
 object N5DataLayer {
   implicit val jsonFormat: OFormat[N5DataLayer] = Json.format[N5DataLayer]
@@ -57,7 +74,26 @@ case class N5SegmentationLayer(
     additionalAxes: Option[Seq[AdditionalAxis]] = None,
     attachments: Option[DatasetLayerAttachments] = None,
 ) extends SegmentationLayer
-    with N5Layer
+    with N5Layer {
+  override def asAbstractLayer: DataLayerLike =
+    AbstractSegmentationLayer(
+      name,
+      category,
+      boundingBox,
+      resolutions,
+      elementClass,
+      largestSegmentId,
+      mappings,
+      defaultViewConfiguration,
+      adminViewConfiguration,
+      coordinateTransformations,
+      additionalAxes,
+      attachments,
+      Some(mags),
+      numChannels,
+      Some(dataFormat)
+    )
+}
 
 object N5SegmentationLayer {
   implicit val jsonFormat: OFormat[N5SegmentationLayer] = Json.format[N5SegmentationLayer]

webknossos-datastore/app/com/scalableminds/webknossos/datastore/dataformats/layers/PrecomputedDataLayers.scala

@@ -37,7 +37,24 @@ case class PrecomputedDataLayer(
     override val numChannels: Option[Int] = Some(1),
     additionalAxes: Option[Seq[AdditionalAxis]] = None,
     attachments: Option[DatasetLayerAttachments] = None,
-) extends PrecomputedLayer
+) extends PrecomputedLayer {
+  override def asAbstractLayer: DataLayerLike =
+    AbstractDataLayer(
+      name,
+      category,
+      boundingBox,
+      resolutions,
+      elementClass,
+      defaultViewConfiguration,
+      adminViewConfiguration,
+      coordinateTransformations,
+      additionalAxes,
+      attachments,
+      Some(mags),
+      numChannels,
+      Some(dataFormat)
+    )
+}
 
 object PrecomputedDataLayer {
   implicit val jsonFormat: OFormat[PrecomputedDataLayer] = Json.format[PrecomputedDataLayer]
@@ -57,7 +74,26 @@ case class PrecomputedSegmentationLayer(
     additionalAxes: Option[Seq[AdditionalAxis]] = None,
     attachments: Option[DatasetLayerAttachments] = None,
 ) extends SegmentationLayer
-    with PrecomputedLayer
+    with PrecomputedLayer {
+  override def asAbstractLayer: DataLayerLike =
+    AbstractSegmentationLayer(
+      name,
+      category,
+      boundingBox,
+      resolutions,
+      elementClass,
+      largestSegmentId,
+      mappings,
+      defaultViewConfiguration,
+      adminViewConfiguration,
+      coordinateTransformations,
+      additionalAxes,
+      attachments,
+      Some(mags),
+      numChannels,
+      Some(dataFormat)
+    )
+}
 
 object PrecomputedSegmentationLayer {
   implicit val jsonFormat: OFormat[PrecomputedSegmentationLayer] = Json.format[PrecomputedSegmentationLayer]