feat(file-service): add model management API and unify shared resource access

common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/DocumentFactory.scala

@@ -60,8 +60,8 @@ object DocumentFactory {
     */
   def openReadonlyDocument(fileUri: URI): ReadonlyVirtualDocument[_] = {
     fileUri.getScheme match {
-      case DATASET_FILE_URI_SCHEME => new DatasetFileDocument(fileUri)
-      case MODEL_FILE_URI_SCHEME   => new ModelFileDocument(fileUri)
+      case DATASET_FILE_URI_SCHEME => new LakeFSFileDocument(fileUri, ResourceType.Datasets)
+      case MODEL_FILE_URI_SCHEME   => new LakeFSFileDocument(fileUri, ResourceType.Models)
       case "file"                  => new ReadonlyLocalFileDocument(fileUri)
       case unsupportedScheme =>
         throw new UnsupportedOperationException(
@@ -184,8 +184,8 @@ object DocumentFactory {
     */
   def openDocument(uri: URI): (VirtualDocument[_], Option[Schema]) = {
     uri.getScheme match {
-      case DATASET_FILE_URI_SCHEME => (new DatasetFileDocument(uri), None)
-      case MODEL_FILE_URI_SCHEME   => (new ModelFileDocument(uri), None)
+      case DATASET_FILE_URI_SCHEME => (new LakeFSFileDocument(uri, ResourceType.Datasets), None)
+      case MODEL_FILE_URI_SCHEME   => (new LakeFSFileDocument(uri, ResourceType.Models), None)
       case VFS_FILE_URI_SCHEME =>
         val IcebergLocation(warehouse, namespace, storageKey) = resolveIcebergLocation(uri)
 

common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/LakeFSFileDocument.scala

@@ -21,6 +21,7 @@ package org.apache.texera.amber.core.storage.model
 
 import com.typesafe.scalalogging.LazyLogging
 import org.apache.texera.common.config.EnvironmentalVariable
+import org.apache.texera.amber.core.storage.ResourceType
 import org.apache.texera.amber.core.storage.model.LakeFSFileDocument.userJwtToken
 import org.apache.texera.amber.core.storage.util.LakeFSStorageClient
 
@@ -36,21 +37,51 @@ object LakeFSFileDocument {
   // In the local development or other architectures, this token can be empty.
   lazy val userJwtToken: String =
     sys.env.getOrElse(EnvironmentalVariable.ENV_USER_JWT_TOKEN, "").trim
+
+  private lazy val datasetPresignEndpoint: String =
+    sys.env
+      .getOrElse(
+        EnvironmentalVariable.ENV_FILE_SERVICE_GET_DATASET_PRESIGNED_URL_ENDPOINT,
+        "http://localhost:9092/api/dataset/presign-download"
+      )
+      .trim
+
+  private lazy val modelPresignEndpoint: String =
+    sys.env
+      .getOrElse(
+        EnvironmentalVariable.ENV_FILE_SERVICE_GET_MODEL_PRESIGNED_URL_ENDPOINT,
+        "http://localhost:9092/api/model/presign-download"
+      )
+      .trim
+
+  /**
+    * The file-service presign-download endpoint serving this resource type. Each resource type
+    * owns an endpoint because they enforce different access control (a dataset grant does not
+    * grant a model).
+    */
+  def presignEndpointOf(resourceType: ResourceType.Value): String =
+    resourceType match {
+      case ResourceType.Datasets => datasetPresignEndpoint
+      case ResourceType.Models   => modelPresignEndpoint
+    }
 }
 
 /**
   * A read-only document over a single file stored in a LakeFS repository, addressed by the URI
-  * {scheme}:///{repositoryName}/{versionHash}/{fileRelativePath}. This is the shared behavior
-  * for every versioned-file resource (datasets, models, …): the file bytes are fetched via a
-  * presigned URL, falling back to a direct LakeFS fetch.
+  * {scheme}:///{repositoryName}/{versionHash}/{fileRelativePath}.
+  *
+  * Every versioned-file resource (datasets, models, …) reads its files the same way — fetch the
+  * bytes through a presigned URL, falling back to a direct LakeFS fetch
   *
-  * @param uri             the resolved {scheme}:///{repositoryName}/{versionHash}/{file} URI
-  * @param presignEndpoint the file-service presign-download endpoint for this resource kind
+  * @param uri          the resolved {scheme}:///{repositoryName}/{versionHash}/{file} URI
+  * @param resourceType which resource this file belongs to, selecting the presign endpoint
   */
-private[storage] abstract class LakeFSFileDocument(uri: URI, presignEndpoint: String)
+private[storage] class LakeFSFileDocument(uri: URI, val resourceType: ResourceType.Value)
     extends VirtualDocument[Nothing]
     with OnVersionedFileResource
     with LazyLogging {
+
+  private val presignEndpoint: String = LakeFSFileDocument.presignEndpointOf(resourceType)
   // Utility function to parse and decode URI segments into individual components
   private def parseUri(uri: URI): (String, String, Path) = {
     val segments = Paths.get(uri.getPath).iterator().asScala.map(_.toString).toArray

common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/DocumentFactorySpec.scala

@@ -20,8 +20,7 @@
 package org.apache.texera.amber.core.storage
 
 import org.apache.texera.amber.core.storage.model.{
-  DatasetFileDocument,
-  ModelFileDocument,
+  LakeFSFileDocument,
   OnVersionedFileResource,
   ReadonlyLocalFileDocument,
   VirtualDocument
@@ -134,17 +133,19 @@ class DocumentFactorySpec extends AnyFlatSpec with Matchers with BeforeAndAfterA
 
   private val versionHash = "97fd4c2a755b69b7c66d322eab40b7e5c2ad5d10"
 
-  "openReadonlyDocument" should "return a DatasetFileDocument for the dataset scheme" in {
+  "openReadonlyDocument" should "return a dataset-typed LakeFSFileDocument for the dataset scheme" in {
     val datasetUri = new URI(s"dataset:///repo/$versionHash/file.txt")
     val doc = DocumentFactory.openReadonlyDocument(datasetUri)
-    doc shouldBe a[DatasetFileDocument]
+    doc shouldBe a[LakeFSFileDocument]
+    doc.asInstanceOf[LakeFSFileDocument].resourceType shouldBe ResourceType.Datasets
     doc.getURI shouldBe datasetUri
   }
 
-  it should "return a ModelFileDocument for the model scheme and parse its URI components" in {
+  it should "return a model-typed LakeFSFileDocument for the model scheme and parse its URI components" in {
     val modelUri = new URI(s"model:///model-1/$versionHash/weights/model.pt")
     val doc = DocumentFactory.openReadonlyDocument(modelUri)
-    doc shouldBe a[ModelFileDocument]
+    doc shouldBe a[LakeFSFileDocument]
+    doc.asInstanceOf[LakeFSFileDocument].resourceType shouldBe ResourceType.Models
     doc.getURI shouldBe modelUri
 
     val resource = doc.asInstanceOf[OnVersionedFileResource]
@@ -176,17 +177,19 @@ class DocumentFactorySpec extends AnyFlatSpec with Matchers with BeforeAndAfterA
   // openDocument / createDocument / documentExists -- unsupported schemes
   // ---------------------------------------------------------------------------
 
-  "openDocument" should "return a DatasetFileDocument and no schema for the dataset scheme" in {
+  "openDocument" should "return a dataset-typed LakeFSFileDocument and no schema for the dataset scheme" in {
     val datasetUri = new URI(s"dataset:///repo/$versionHash/file.txt")
     val (doc, schemaOpt) = DocumentFactory.openDocument(datasetUri)
-    doc shouldBe a[DatasetFileDocument]
+    doc shouldBe a[LakeFSFileDocument]
+    doc.asInstanceOf[LakeFSFileDocument].resourceType shouldBe ResourceType.Datasets
     schemaOpt shouldBe None
   }
 
-  it should "return a ModelFileDocument and no schema for the model scheme" in {
+  it should "return a model-typed LakeFSFileDocument and no schema for the model scheme" in {
     val modelUri = new URI(s"model:///model-1/$versionHash/weights/model.pt")
     val (doc, schemaOpt) = DocumentFactory.openDocument(modelUri)
-    doc shouldBe a[ModelFileDocument]
+    doc shouldBe a[LakeFSFileDocument]
+    doc.asInstanceOf[LakeFSFileDocument].resourceType shouldBe ResourceType.Models
     schemaOpt shouldBe None
   }