feat(dataset): add contributor metadata storage and API

file-service/src/main/scala/org/apache/texera/service/resource/DatasetResource.scala

@@ -34,6 +34,7 @@ import org.apache.texera.dao.SqlServer
 import org.apache.texera.dao.SqlServer.withTransaction
 import org.apache.texera.dao.jooq.generated.enums.PrivilegeEnum
 import org.apache.texera.dao.jooq.generated.tables.Dataset.DATASET
+import org.apache.texera.dao.jooq.generated.tables.DatasetContributor.DATASET_CONTRIBUTOR
 import org.apache.texera.dao.jooq.generated.tables.DatasetUserAccess.DATASET_USER_ACCESS
 import org.apache.texera.dao.jooq.generated.tables.DatasetVersion.DATASET_VERSION
 import org.apache.texera.dao.jooq.generated.tables.User.USER
@@ -175,12 +176,74 @@ object DatasetResource {
     normalized
   }
 
+  /**
+    * Helper function to get the contributors using the did
+    */
+  def getContributorsByDid(ctx: DSLContext, did: Integer): List[Contributor] = {
+    ctx
+      .selectFrom(DATASET_CONTRIBUTOR)
+      .where(DATASET_CONTRIBUTOR.DID.eq(did))
+      .fetch()
+      .asScala
+      .toList
+      .map { record =>
+        Contributor(
+          name = record.getName,
+          creator = record.getCreator,
+          affiliation = record.getAffiliation,
+          email = record.getEmail,
+          comments = record.getComments
+        )
+      }
+  }
+
+  /**
+    * Helper function to insert the contributors of a dataset in one batch
+    */
+  def insertContributors(ctx: DSLContext, did: Integer, contributors: List[Contributor]): Unit = {
+    val records = contributors.map { contributor =>
+      if (contributor == null || contributor.name == null || contributor.name.trim.isEmpty) {
+        throw new BadRequestException("Each contributor must have a name")
+      }
+      if (
+        contributor.name.length > 256 ||
+        Option(contributor.email).exists(_.length > 256) ||
+        Option(contributor.affiliation).exists(_.length > 256)
+      ) {
+        throw new BadRequestException("Contributor fields must not exceed 256 characters")
+      }
+      val record = ctx.newRecord(DATASET_CONTRIBUTOR)
+      record.setDid(did)
+      record.setName(contributor.name)
+      record.setCreator(contributor.creator)
+      record.setAffiliation(contributor.affiliation)
+      record.setEmail(contributor.email)
+      record.setComments(contributor.comments)
+      record
+    }
+    ctx.batchInsert(records.asJava).execute()
+  }
+
+  case class Contributor(
+      name: String,
+      creator: Boolean = false,
+      affiliation: String = null,
+      email: String = null,
+      comments: String = null
+  )
+
+  case class DatasetContributorsModification(
+      did: Integer,
+      contributors: Option[List[Contributor]] = None
+  )
+
   case class DashboardDataset(
       dataset: Dataset,
       ownerEmail: String,
       accessPrivilege: EnumType,
       isOwner: Boolean,
-      size: Long
+      size: Long,
+      contributors: List[Contributor] = Nil
   )
 
   case class DashboardDatasetVersion(
@@ -192,7 +255,8 @@ object DatasetResource {
       datasetName: String,
       datasetDescription: String,
       isDatasetPublic: Boolean,
-      isDatasetDownloadable: Boolean
+      isDatasetDownloadable: Boolean,
+      contributors: Option[List[Contributor]] = None
   )
 
   case class Diff(
@@ -257,7 +321,8 @@ class DatasetResource extends LazyLogging {
       isOwner,
       withLakeFSErrorHandling(s"retrieving the size of dataset '${targetDataset.getName}'") {
         LakeFSStorageClient.retrieveRepositorySize(targetDataset.getRepositoryName)
-      }
+      },
+      contributors = DatasetResource.getContributorsByDid(ctx, did)
     )
   }
 
@@ -309,6 +374,9 @@ class DatasetResource extends LazyLogging {
           .fetchOne()
       }
 
+      val savedContributors = request.contributors.getOrElse(Nil)
+      DatasetResource.insertContributors(ctx, createdDataset.getDid, savedContributors)
+
       // Initialize the repository in LakeFS
       val repositoryName = s"dataset-${createdDataset.getDid}"
       try {
@@ -347,7 +415,8 @@ class DatasetResource extends LazyLogging {
         user.getEmail,
         PrivilegeEnum.WRITE,
         isOwner = true,
-        0
+        0,
+        savedContributors
       )
     }
   }
@@ -494,6 +563,34 @@ class DatasetResource extends LazyLogging {
     }
   }
 
+  @POST
+  @Consumes(Array(MediaType.APPLICATION_JSON))
+  @Produces(Array(MediaType.APPLICATION_JSON))
+  @RolesAllowed(Array("REGULAR", "ADMIN"))
+  @Path("/update/contributors")
+  def updateDatasetContributors(
+      modificator: DatasetContributorsModification,
+      @Auth user: SessionUser
+  ): Response = {
+    withTransaction(context) { ctx =>
+      if (!userHasWriteAccess(ctx, modificator.did, user.getUid)) {
+        throw new ForbiddenException(ERR_USER_HAS_NO_ACCESS_TO_DATASET_MESSAGE)
+      }
+
+      ctx
+        .delete(DATASET_CONTRIBUTOR)
+        .where(DATASET_CONTRIBUTOR.DID.eq(modificator.did))
+        .execute()
+      DatasetResource.insertContributors(
+        ctx,
+        modificator.did,
+        modificator.contributors.getOrElse(Nil)
+      )
+
+      Response.ok().build()
+    }
+  }
+
   @POST
   @Consumes(Array(MediaType.APPLICATION_JSON))
   @Produces(Array(MediaType.APPLICATION_JSON))

file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala

@@ -343,6 +343,40 @@ class DatasetResourceSpec
     dashboardDataset.dataset.getIsDownloadable shouldBe false
   }
 
+  it should "persist contributors and return them in the response" in {
+    val contributors = List(
+      DatasetResource.Contributor(
+        name = "test1",
+        creator = true,
+        affiliation = "Test Lab A",
+        email = "contributor-a@test.com",
+        comments = "collected the data"
+      ),
+      DatasetResource.Contributor(
+        name = "test2",
+        creator = false,
+        affiliation = "Test Lab B",
+        email = "contributor-b@test.com",
+        comments = null
+      )
+    )
+    val createDatasetRequest = DatasetResource.CreateDatasetRequest(
+      datasetName = "contributor-ds",
+      datasetDescription = "dataset with contributors",
+      isDatasetPublic = false,
+      isDatasetDownloadable = true,
+      contributors = Some(contributors)
+    )
+
+    val createdDataset = datasetResource.createDataset(createDatasetRequest, sessionUser)
+
+    createdDataset.contributors should contain theSameElementsAs contributors
+    DatasetResource.getContributorsByDid(
+      getDSLContext,
+      createdDataset.dataset.getDid
+    ) should contain theSameElementsAs contributors
+  }
+
   it should "delete dataset successfully if user owns it" in {
     val dataset = new Dataset
     dataset.setName("delete-ds")
@@ -404,6 +438,197 @@ class DatasetResourceSpec
     dashboardDataset.size should be >= 0L
   }
 
+  "updateDatasetContributors" should "replace the contributor list of the dataset" in {
+    val initial = List(
+      DatasetResource.Contributor(
+        "test1",
+        creator = true,
+        "Test Lab A",
+        "contributor-a@test.com",
+        "initial"
+      )
+    )
+    val createdDataset = datasetResource.createDataset(
+      DatasetResource.CreateDatasetRequest(
+        datasetName = "contributor-update-ds",
+        datasetDescription = "dataset for contributor update",
+        isDatasetPublic = false,
+        isDatasetDownloadable = true,
+        contributors = Some(initial)
+      ),
+      sessionUser
+    )
+    val did = createdDataset.dataset.getDid
+
+    val replacement = List(
+      DatasetResource
+        .Contributor("test2", creator = false, "Test Lab C", "contributor-c@test.com", "curation"),
+      DatasetResource.Contributor(
+        "test3",
+        creator = true,
+        "Test Lab D",
+        "contributor-d@test.com",
+        "analysis"
+      )
+    )
+    val response = datasetResource.updateDatasetContributors(
+      DatasetResource.DatasetContributorsModification(did, Some(replacement)),
+      sessionUser
+    )
+
+    response.getStatus shouldEqual 200
+    DatasetResource.getContributorsByDid(
+      getDSLContext,
+      did
+    ) should contain theSameElementsAs replacement
+  }
+
+  it should "clear all contributors when given an empty list" in {
+    val createdDataset = datasetResource.createDataset(
+      DatasetResource.CreateDatasetRequest(
+        datasetName = "contributor-clear-ds",
+        datasetDescription = "dataset for contributor clear test",
+        isDatasetPublic = false,
+        isDatasetDownloadable = true,
+        contributors = Some(
+          List(
+            DatasetResource
+              .Contributor("test1", creator = true, "Test Lab A", "contributor-a@test.com", null),
+            DatasetResource
+              .Contributor("test2", creator = false, "Test Lab B", "contributor-b@test.com", null)
+          )
+        )
+      ),
+      sessionUser
+    )
+    val did = createdDataset.dataset.getDid
+    DatasetResource.getContributorsByDid(getDSLContext, did) should have size 2
+
+    val response = datasetResource.updateDatasetContributors(
+      DatasetResource.DatasetContributorsModification(did, Some(Nil)),
+      sessionUser
+    )
+
+    response.getStatus shouldEqual 200
+    DatasetResource.getContributorsByDid(getDSLContext, did) shouldBe empty
+  }
+
+  it should "reject a contributor without a name" in {
+    val createdDataset = datasetResource.createDataset(
+      DatasetResource.CreateDatasetRequest(
+        datasetName = "contributor-noname-ds",
+        datasetDescription = "dataset for contributor validation test",
+        isDatasetPublic = false,
+        isDatasetDownloadable = true
+      ),
+      sessionUser
+    )
+    val did = createdDataset.dataset.getDid
+
+    assertThrows[BadRequestException] {
+      datasetResource.updateDatasetContributors(
+        DatasetResource.DatasetContributorsModification(
+          did,
+          Some(
+            List(
+              DatasetResource
+                .Contributor(null, creator = false, "Test Lab A", "contributor-x@test.com", null)
+            )
+          )
+        ),
+        sessionUser
+      )
+    }
+    DatasetResource.getContributorsByDid(getDSLContext, did) shouldBe empty
+  }
+
+  it should "reject contributor fields longer than 256 characters" in {
+    val createdDataset = datasetResource.createDataset(
+      DatasetResource.CreateDatasetRequest(
+        datasetName = "contributor-toolong-ds",
+        datasetDescription = "dataset for contributor length test",
+        isDatasetPublic = false,
+        isDatasetDownloadable = true
+      ),
+      sessionUser
+    )
+
+    assertThrows[BadRequestException] {
+      datasetResource.updateDatasetContributors(
+        DatasetResource.DatasetContributorsModification(
+          createdDataset.dataset.getDid,
+          Some(
+            List(
+              DatasetResource
+                .Contributor(
+                  "A" * 257,
+                  creator = false,
+                  "Test Lab A",
+                  "contributor-x@test.com",
+                  null
+                )
+            )
+          )
+        ),
+        sessionUser
+      )
+    }
+  }
+
+  it should "refuse to update contributors without write access" in {
+    val createdDataset = datasetResource.createDataset(
+      DatasetResource.CreateDatasetRequest(
+        datasetName = "contributor-forbidden-ds",
+        datasetDescription = "dataset for contributor permission test",
+        isDatasetPublic = true,
+        isDatasetDownloadable = true
+      ),
+      sessionUser
+    )
+    val did = createdDataset.dataset.getDid
+
+    assertThrows[ForbiddenException] {
+      datasetResource.updateDatasetContributors(
+        DatasetResource.DatasetContributorsModification(
+          did,
+          Some(
+            List(
+              DatasetResource
+                .Contributor("test1", creator = false, "Test Lab E", "contributor-e@test.com", null)
+            )
+          )
+        ),
+        multipartNoWriteSessionUser
+      )
+    }
+
+    DatasetResource.getContributorsByDid(getDSLContext, did) shouldBe empty
+  }
+
+  it should "remove contributors when their dataset is deleted" in {
+    val createdDataset = datasetResource.createDataset(
+      DatasetResource.CreateDatasetRequest(
+        datasetName = "contributor-cascade-ds",
+        datasetDescription = "dataset for contributor cascade test",
+        isDatasetPublic = false,
+        isDatasetDownloadable = true,
+        contributors = Some(
+          List(
+            DatasetResource
+              .Contributor("test1", creator = true, "Test Lab A", "contributor-f@test.com", null)
+          )
+        )
+      ),
+      sessionUser
+    )
+    val did = createdDataset.dataset.getDid
+    DatasetResource.getContributorsByDid(getDSLContext, did) should have size 1
+
+    datasetResource.deleteDataset(did, sessionUser).getStatus shouldEqual 200
+
+    DatasetResource.getContributorsByDid(getDSLContext, did) shouldBe empty
+  }
+
   "findExistingUploadFiles" should "match committed and staged files by path and size" in {
     val repoName = s"existing-upload-${System.nanoTime()}"
     val dataset = new Dataset