Add s3 storage path interfaces

framework/arcane-framework/build.sbt

@@ -5,5 +5,20 @@ ThisBuild / scalaVersion := "3.6.1"
 lazy val root = (project in file("."))
   .settings(
     name := "arcane-framework",
-    idePackagePrefix := Some("com.sneaksanddata.arcane.framework")
+    idePackagePrefix := Some("com.sneaksanddata.arcane.framework"),
+
+    // Compiler options
+    Test / logBuffered := false,
+
+    // Framework dependencies
+    libraryDependencies += "io.delta" % "delta-kernel-api" % "4.0.0rc1",
+    libraryDependencies += "dev.zio" %% "zio" % "2.1.6",
+    libraryDependencies += "dev.zio" %% "zio-streams" % "2.1.6",
+    libraryDependencies += "com.microsoft.sqlserver" % "mssql-jdbc" % "12.8.1.jre11",
+    libraryDependencies += "software.amazon.awssdk" % "s3" % "2.25.27",
+
+    // Test dependencies
+    libraryDependencies += "org.scalatest" %% "scalatest" % "3.2.19" % Test,
+    libraryDependencies += "org.scalatest" %% "scalatest-flatspec" % "3.2.19" % Test
+
   )

framework/arcane-framework/src/main/scala/services/storage/base/BlobStorageWriter.scala

@@ -0,0 +1,52 @@
+package com.sneaksanddata.arcane.framework
+package services.storage.base
+
+import services.storage.exceptions.BlobStorageException
+import services.storage.models.UploadResult
+import services.storage.models.base.BlobPath
+
+import zio.ZIO
+
+import java.net.URL
+
+/**
+ * A trait that defines the interface for writing to a blob storage.
+ *
+ * @tparam PathType The type of the path to the blob.
+ */
+trait BlobStorageWriter[PathType <: BlobPath] {
+  /**
+   * Saves the given bytes as a blob.
+   *
+   * @param blobPath The path to the blob.
+   * @param data The bytes to save.
+   * @return The result of the upload.
+   */
+  def saveBytesAsBlob(blobPath: PathType, data: Array[Byte]): ZIO[Any, BlobStorageException, UploadResult]
+
+  /**
+   * Saves the given text as a blob.
+   *
+   * @param blobPath The path to the blob.
+   * @param data The text to save.
+   * @return The result of the upload.
+   */
+  def saveTextAsBlob(blobPath: PathType, data: String): ZIO[Any, BlobStorageException, UploadResult]
+
+  /**
+   * Removes the blob at the given path.
+   *
+   * @param blobPath The path to the blob.
+   * @param data The data to remove.
+   */
+  def removeBlob(blobPath: PathType, data: String): ZIO[Any, BlobStorageException, Unit]
+
+  /**
+   * Gets the URI of the blob at the given path.
+   *
+   * @param blobPath The path to the blob.
+   * @param data The data to get.
+   * @return The URI of the blob.
+   */
+  def getBlobUri(blobPath: PathType, data: String): ZIO[Any, BlobStorageException, URL]
+}

framework/arcane-framework/src/main/scala/services/storage/base/StorageBlobReader.scala

@@ -0,0 +1,24 @@
+package com.sneaksanddata.arcane.framework
+package services.storage.base
+
+import services.storage.exceptions.BlobStorageException
+import services.storage.models.base.BlobPath
+
+import zio.ZIO
+
+/**
+ * A trait that defines the interface for reading from a blob storage.
+ *
+ * @tparam PathType The type of the path to the blob.
+ */
+trait StorageBlobReader[PathType <: BlobPath] {
+  /**
+   * Gets the content of the blob at the given path.
+   *
+   * @param blobPath The path to the blob.
+   * @param deserializer function to deserialize the content of the blob.
+   * @tparam Result The type of the result.
+   * @return The result of applying the function to the content of the blob.
+   */
+  def getBlobContent[Result](blobPath: PathType, deserializer: Array[Byte] => Result): ZIO[Any, BlobStorageException, Result]
+}

framework/arcane-framework/src/main/scala/services/storage/exceptions/BlobStorageException.scala

@@ -0,0 +1,10 @@
+package com.sneaksanddata.arcane.framework
+package services.storage.exceptions
+
+/**
+ * An exception that is thrown when an error occurs in the blob storage.
+ *
+ * @param message The message of the exception.
+ * @param cause The cause of the exception.
+ */
+final case class BlobStorageException(private val message: String, private val cause: Throwable) extends Exception(message, cause);

framework/arcane-framework/src/main/scala/services/storage/models/UploadResult.scala

@@ -0,0 +1,13 @@
+package com.sneaksanddata.arcane.framework
+package services.storage.models
+
+import java.time.OffsetDateTime
+
+/**
+ * The result of uploading a blob to a storage.
+ *
+ * @param name The name of the blob.
+ * @param lastModified The last modified time of the blob.
+ * @param contentHash The hash of the content of the blob.
+ */
+final case class UploadResult(name: String, lastModified: OffsetDateTime, contentHash: String)

framework/arcane-framework/src/main/scala/services/storage/models/amazon/AmazonS3StoragePath.scala

@@ -0,0 +1,62 @@
+package com.sneaksanddata.arcane.framework
+package services.storage.models.amazon
+
+import services.storage.models.base.BlobPath
+
+import scala.util.matching.Regex
+import scala.util.{Failure, Success, Try}
+
+/**
+  * Represents a path to a blob in Amazon S3 storage.
+  *
+  * @param bucket The name of the bucket.
+  * @param objectKey The key of the object in the bucket.
+  */
+private final case class AmazonS3StoragePath(bucket: String, objectKey: String) extends BlobPath {
+
+  /**
+   * Converts the path to a HDFS-style path.
+   *
+   * @return The path as a string.
+   */
+  override def toHdfsPath = s"s3a://$bucket/$objectKey"
+
+  /**
+    * Joins the given key name to the current path.
+    *
+    * @param keyName The key name to join.
+    * @return The new path.
+    */
+  def join(keyName: String) = new AmazonS3StoragePath(bucket, if (objectKey.isEmpty) keyName else s"$objectKey/$keyName")
+
+}
+
+/**
+  * Companion object for [[AmazonS3StoragePath]].
+  */
+object AmazonS3StoragePath {
+  private val matchRegex: String = "s3a://([^/]+)/?(.*)"
+
+  /**
+    * Checks if the given path is an Amazon S3 path.
+    *
+    * @param hdfsPath The path to check.
+    * @return True if the path is an Amazon S3 path, false otherwise.
+    */
+  def isAmazonS3Path(hdfsPath: String): Boolean = hdfsPath.matches(matchRegex)
+
+  /**
+    * Creates an [[AmazonS3StoragePath]] from the given HDFS path.
+    *
+    * @param hdfsPath The HDFS path.
+    * @return The [[AmazonS3StoragePath]].
+    */
+  def fromHdfsPath(hdfsPath: String): Try[AmazonS3StoragePath] = {
+    val r: Regex = AmazonS3StoragePath.matchRegex.r
+    val m = r.findFirstMatchIn(hdfsPath)
+    m match {
+      case Some(matched) => Success(AmazonS3StoragePath(matched.group(1), matched.group(2).stripSuffix("/")))
+      case None => Failure(IllegalArgumentException(s"An AmazonS3StoragePath must be in the format s3a://bucket/path, but was: $hdfsPath"))
+    }
+  }
+}

framework/arcane-framework/src/main/scala/services/storage/models/base/BlobPath.scala

@@ -0,0 +1,14 @@
+package com.sneaksanddata.arcane.framework
+package services.storage.models.base
+
+/**
+ * A trait that represents a path to a blob.
+ */
+trait BlobPath {
+  /**
+   * Converts the path to a HDFS-style path.
+   *
+   * @return The path as a string.
+   */
+  def toHdfsPath: String
+}

framework/arcane-framework/src/test/scala/services/storage/models/amazon/AmazonS3StoragePath.scala

@@ -0,0 +1,56 @@
+import com.sneaksanddata.arcane.framework.services.storage.models.amazon.AmazonS3StoragePath
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.prop.TableDrivenPropertyChecks.forAll
+import org.scalatest.prop.Tables.Table
+
+class AmazonS3StoragePath extends AnyFlatSpec {
+
+  "AmazonS3StoragePath" should "be able to parse correct path" in {
+    val path = "s3a://bucket/key"
+    val parsed = AmazonS3StoragePath.fromHdfsPath(path)
+    assert(parsed.isSuccess)
+    assert(parsed.get.bucket == "bucket")
+    assert(parsed.get.objectKey == "key")
+  }
+
+  it should "have stable serialization and deserialization" in {
+    val path = "s3a://bucket/key"
+    val parsed = AmazonS3StoragePath.fromHdfsPath(path)
+    val serialized = AmazonS3StoragePath.fromHdfsPath(parsed.get.toHdfsPath)
+
+    assert(parsed.isSuccess)
+    assert(serialized.isSuccess)
+    assert(parsed.get.bucket == serialized.get.bucket)
+    assert(parsed.get.objectKey == serialized.get.objectKey)
+    assert(parsed.get.toHdfsPath == path)
+  }
+
+  it should "be able to jon paths" in {
+    val path = "s3a://bucket/key"
+    val parsed = AmazonS3StoragePath.fromHdfsPath(path)
+    assert(parsed.isSuccess)
+
+    val parsedJoined = parsed.get.join("key2")
+    assert(parsed.get.bucket == "bucket")
+    assert(parsedJoined.objectKey == "key/key2")
+  }
+
+
+  val cases = Table(
+    ("original", "joined", "expected"),  // First tuple defines column names
+    ("s3a://bucket-name/", "/folder1///folder2/file.txt", "s3a://bucket-name//folder1///folder2/file.txt"), // Subsequent tuples define the data
+    ("s3a://bucket-name/", "folder1///folder2/file.txt", "s3a://bucket-name/folder1///folder2/file.txt"),
+    ("s3a://bucket-name", "folder1///folder2/file.txt", "s3a://bucket-name/folder1///folder2/file.txt"),
+    ("s3a://bucket-name", "/folder1///folder2/file.txt", "s3a://bucket-name//folder1///folder2/file.txt")
+  )
+
+  forAll (cases) { (orig: String, rest: String, expected: String ) =>
+    it should f"be able to remove extra slashes with values ($orig, $rest, $expected)" in {
+      val parsed = AmazonS3StoragePath.fromHdfsPath(orig)
+      assert(parsed.isSuccess)
+      val result = parsed.get.join(rest)
+
+      assert(result.toHdfsPath == expected)
+    }
+  }
+}