objectstorex

defmodule ObjectStoreX do
  @moduledoc """
  Unified object storage for AWS S3, Azure Blob Storage, Google Cloud Storage,
  and local filesystem.

  ObjectStoreX provides a consistent API across multiple cloud storage providers,
  powered by the battle-tested Rust [object_store](https://docs.rs/object_store)
  library via Rustler NIFs for near-native performance.

  ## Features

  - **Multi-provider**: S3, Azure, GCS, Local filesystem, In-memory
  - **Streaming**: Efficient large file uploads/downloads with `ObjectStoreX.Stream`
  - **Atomic operations**: Create-only, Compare-And-Swap (CAS), conditional operations
  - **HTTP-style caching**: ETag-based conditional requests (If-Match, If-None-Match)
  - **Bulk operations**: Delete many objects efficiently with automatic batching
  - **Range reads**: Fetch specific byte ranges without downloading entire files
  - **Rich metadata**: Content-type, custom metadata, cache control, tags
  - **Production-ready**: Comprehensive error handling with retryable error detection

  ## Installation

  Add `objectstorex` to your `mix.exs`:

      def deps do
        [
          {:objectstorex, "~> 0.1.0"}
        ]
      end

  ## Quick Start

      # Create a store (S3 example)
      {:ok, store} = ObjectStoreX.new(:s3,
        bucket: "my-bucket",
        region: "us-east-1"
      )

      # Upload an object
      :ok = ObjectStoreX.put(store, "hello.txt", "Hello, World!")

      # Download an object
      {:ok, data} = ObjectStoreX.get(store, "hello.txt")

      # List objects with streaming
      ObjectStoreX.Stream.list_stream(store, prefix: "data/")
      |> Enum.take(10)

      # Delete an object
      :ok = ObjectStoreX.delete(store, "hello.txt")

  ## Supported Providers

  - **`:s3`** - Amazon S3, MinIO, Cloudflare R2, and S3-compatible services
  - **`:azure`** - Azure Blob Storage
  - **`:gcs`** - Google Cloud Storage
  - **`:local`** - Local filesystem
  - **`:memory`** - In-memory storage (for testing)

  See `new/2` for provider-specific configuration options.

  ## Use Cases

  ### File Storage
  Upload and download files to cloud storage with automatic retries and error handling.

  ### Data Lakes
  Store analytics data with partitioning and efficient listing:

      ObjectStoreX.Stream.list_stream(store, prefix: "data/2025/01/")
      |> Stream.filter(fn meta -> meta.size > 1_000_000 end)
      |> Enum.to_list()

  ### Distributed Locks
  Use create-only writes for coordination:

      case ObjectStoreX.put(store, "lock.txt", "locked", mode: :create) do
        {:ok, _meta} -> :acquired
        {:error, :already_exists} -> :already_locked
      end

  ### HTTP Caching
  ETag-based cache validation:

      case ObjectStoreX.get(store, "file.txt", if_none_match: cached_etag) do
        {:ok, data, meta} -> {:modified, data, meta}
        {:error, :not_modified} -> :use_cache
      end

  ### Optimistic Locking
  Compare-and-swap for concurrent updates:

      {:ok, meta} = ObjectStoreX.head(store, "counter.json")
      case ObjectStoreX.put(store, "counter.json", new_data,
                            mode: {:update, %{etag: meta[:etag]}}) do
        {:ok, _meta} -> :success
        {:error, :precondition_failed} -> :retry
      end

  ### Large File Handling
  Stream large files without loading into memory:

      File.stream!("large-file.bin", [], 10_485_760)  # 10MB chunks
      |> ObjectStoreX.Stream.upload(store, "backup.bin")

  ## Error Handling

  All functions return tagged tuples for pattern matching:
  - `{:ok, result}` on success
  - `{:error, reason}` on failure

  Error reasons are descriptive atoms:

      case ObjectStoreX.get(store, "file.txt") do
        {:ok, data} -> process(data)
        {:error, :not_found} -> :missing
        {:error, :permission_denied} -> :unauthorized
        {:error, :timeout} -> :retry_later
      end

  For detailed information about error types, context, and retry strategies,
  see `ObjectStoreX.Error`.

  ## Architecture

  ObjectStoreX uses Rustler NIFs to call the Rust `object_store` library,
  providing near-native performance with memory-safe operations. The Rust
  library handles all provider-specific protocols and optimizations.

  ## Documentation

  - Getting Started: See `guides/getting_started.md`
  - Configuration: See `guides/configuration.md`
  - Streaming: See `ObjectStoreX.Stream` and `guides/streaming.md`
  - Error Handling: See `ObjectStoreX.Error` and `guides/error_handling.md`
  - Distributed Systems: See `guides/distributed_systems.md`

  ## Links

  - [Hex.pm](https://hex.pm/packages/objectstorex)
  - [Documentation](https://hexdocs.pm/objectstorex)
  - [GitHub](https://github.com/yourorg/objectstorex)
  - [Changelog](https://github.com/yourorg/objectstorex/blob/main/CHANGELOG.md)
  """

  alias ObjectStoreX.Native

  @type store :: reference()
  @type path :: String.t()
  @type provider :: :s3 | :azure | :gcs | :local | :memory
  @type metadata :: %{
          location: String.t(),
          last_modified: String.t(),
          size: non_neg_integer(),
          etag: String.t() | nil
        }

  @doc """
  Create a new storage provider.

  ## Examples

      # S3
      {:ok, store} = ObjectStoreX.new(:s3,
        bucket: "my-bucket",
        region: "us-east-1",
        access_key_id: System.get_env("AWS_ACCESS_KEY_ID"),
        secret_access_key: System.get_env("AWS_SECRET_ACCESS_KEY")
      )

      # S3-compatible (Wasabi, MinIO, Cloudflare R2)
      {:ok, store} = ObjectStoreX.new(:s3,
        bucket: "my-bucket",
        region: "eu-central-1",
        endpoint: "https://s3.eu-central-1.wasabisys.com",
        access_key_id: System.get_env("AWS_ACCESS_KEY_ID"),
        secret_access_key: System.get_env("AWS_SECRET_ACCESS_KEY")
      )

      # Azure
      {:ok, store} = ObjectStoreX.new(:azure,
        account: "myaccount",
        container: "mycontainer",
        access_key: System.get_env("AZURE_STORAGE_KEY")
      )

      # GCS
      {:ok, store} = ObjectStoreX.new(:gcs,
        bucket: "my-gcs-bucket",
        service_account_key: File.read!("credentials.json")
      )

      # Local filesystem
      {:ok, store} = ObjectStoreX.new(:local, path: "/tmp/storage")

      # In-memory (for testing)
      {:ok, store} = ObjectStoreX.new(:memory)
  """
  @spec new(provider(), keyword()) :: {:ok, store()} | {:error, term()}
  @spec new(provider()) :: {:ok, store()} | {:error, term()}

  def new(:s3, opts) do
    bucket = Keyword.fetch!(opts, :bucket)
    region = Keyword.get(opts, :region)
    access_key_id = Keyword.get(opts, :access_key_id)
    secret_access_key = Keyword.get(opts, :secret_access_key)
    endpoint = Keyword.get(opts, :endpoint)

    case Native.new_s3(bucket, region, access_key_id, secret_access_key, endpoint) do
      store when is_reference(store) -> {:ok, store}
      error -> {:error, error}
    end
  rescue
    e -> {:error, Exception.message(e)}
  end

  def new(:azure, opts) do
    account = Keyword.fetch!(opts, :account)
    container = Keyword.fetch!(opts, :container)
    access_key = Keyword.get(opts, :access_key)

    case Native.new_azure(account, container, access_key) do
      store when is_reference(store) -> {:ok, store}
      error -> {:error, error}
    end
  rescue
    e -> {:error, Exception.message(e)}
  end

  def new(:gcs, opts) do
    bucket = Keyword.fetch!(opts, :bucket)
    service_account_key = Keyword.get(opts, :service_account_key)

    case Native.new_gcs(bucket, service_account_key) do
      store when is_reference(store) -> {:ok, store}
      error -> {:error, error}
    end
  rescue
    e -> {:error, Exception.message(e)}
  end

  def new(:local, opts) do
    path = Keyword.fetch!(opts, :path)

    case Native.new_local(path) do
      store when is_reference(store) -> {:ok, store}
      error -> {:error, error}
    end
  rescue
    e -> {:error, Exception.message(e)}
  end

  @doc """
  Create an in-memory storage provider (shorthand for testing).

  This is a convenience function equivalent to `new(:memory, [])`.

  ## Examples

      {:ok, store} = ObjectStoreX.new(:memory)
  """
  @spec new(:memory) :: {:ok, store()} | {:error, term()}
  def new(:memory) do
    case Native.new_memory() do
      store when is_reference(store) -> {:ok, store}
      error -> {:error, error}
    end
  rescue
    e -> {:error, Exception.message(e)}
  end

  @type put_result :: %{
          etag: String.t(),
          version: String.t()
        }

  @doc """
  Upload an object to storage.

  ## Options

  - `:mode` - Write mode (default: `:overwrite`)
    - `:overwrite` - Always write, overwriting any existing object
    - `:create` - Only write if the object doesn't exist (returns `{:error, :already_exists}` if exists)
    - `{:update, %{etag: ..., version: ...}}` - CAS operation, only write if version matches
  - `:content_type` - MIME type (e.g., "application/json")
  - `:content_encoding` - Encoding (e.g., "gzip")
  - `:content_disposition` - Download behavior (e.g., "attachment; filename=file.pdf")
  - `:cache_control` - Cache directives (e.g., "max-age=3600")
  - `:content_language` - Language (e.g., "en-US")
  - `:tags` - Object tags as a map (AWS/GCS only)

  ## Examples

      # Simple put (overwrite)
      :ok = ObjectStoreX.put(store, "file.txt", "Hello, World!")

      # Create only (distributed lock)
      case ObjectStoreX.put(store, "lock.txt", "locked", mode: :create) do
        {:ok, _meta} -> :acquired
        {:error, :already_exists} -> :already_locked
      end

      # Compare-and-swap (optimistic locking)
      {:ok, meta} = ObjectStoreX.head(store, "counter.json")
      case ObjectStoreX.put(store, "counter.json", new_data,
                            mode: {:update, %{etag: meta[:etag], version: meta[:version]}}) do
        {:ok, _meta} -> :success
        {:error, :precondition_failed} -> :retry
      end

      # Upload with content type
      ObjectStoreX.put(store, "data.json", json_data, content_type: "application/json")

      # Upload with attributes
      ObjectStoreX.put(store, "report.pdf", pdf_data,
        content_type: "application/pdf",
        content_disposition: "attachment; filename=report.pdf",
        cache_control: "max-age=3600"
      )

      # Upload with tags (AWS/GCS)
      ObjectStoreX.put(store, "backup.zip", data,
        tags: %{"environment" => "production", "backup-type" => "daily"}
      )
  """
  @spec put(store(), path(), binary(), keyword()) ::
          :ok | {:ok, put_result()} | {:error, term()}
  def put(store, path, data, opts \\ [])

  def put(store, path, data, []) when is_binary(data) do
    case Native.put(store, path, data) do
      :ok -> :ok
      error -> {:error, error}
    end
  rescue
    e -> {:error, Exception.message(e)}
  end

  def put(store, path, data, opts) when is_binary(data) and is_list(opts) do
    mode = Keyword.get(opts, :mode, :overwrite)

    result =
      if has_attributes?(opts) do
        put_with_attributes_internal(store, path, data, mode, opts)
      else
        Native.put_with_mode(store, path, data, mode)
      end

    normalize_put_result(result)
  rescue
    e -> {:error, Exception.message(e)}
  end

  defp has_attributes?(opts) do
    Keyword.has_key?(opts, :content_type) or
      Keyword.has_key?(opts, :content_encoding) or
      Keyword.has_key?(opts, :content_disposition) or
      Keyword.has_key?(opts, :cache_control) or
      Keyword.has_key?(opts, :content_language) or
      Keyword.has_key?(opts, :tags)
  end

  defp put_with_attributes_internal(store, path, data, mode, opts) do
    attributes = %ObjectStoreX.Attributes{
      content_type: Keyword.get(opts, :content_type),
      content_encoding: Keyword.get(opts, :content_encoding),
      content_disposition: Keyword.get(opts, :content_disposition),
      cache_control: Keyword.get(opts, :cache_control),
      content_language: Keyword.get(opts, :content_language)
    }

    tags =
      Keyword.get(opts, :tags, %{})
      |> Map.to_list()

    Native.put_with_attributes(store, path, data, mode, attributes, tags)
  end

  defp normalize_put_result({:ok, etag, version}), do: {:ok, %{etag: etag, version: version}}
  defp normalize_put_result(:already_exists), do: {:error, :already_exists}
  defp normalize_put_result(:precondition_failed), do: {:error, :precondition_failed}
  defp normalize_put_result(error), do: {:error, error}

  @doc """
  Download an object from storage with optional conditional requests.

  ## Options

  Supports HTTP-style conditional requests for caching and consistency:
  - `:if_match` - Only return if ETag matches (HTTP If-Match)
  - `:if_none_match` - Only return if ETag differs (HTTP If-None-Match)
  - `:if_modified_since` - Only return if modified after date (DateTime or Unix timestamp)
  - `:if_unmodified_since` - Only return if not modified since date (DateTime or Unix timestamp)
  - `:range` - Byte range `{start, end}` or `%ObjectStoreX.Range{}`
  - `:version` - Specific object version
  - `:head` - Return metadata only (no content)

  ## Examples

      # Simple get
      {:ok, data} = ObjectStoreX.get(store, "file.txt")

      # HTTP cache validation
      case ObjectStoreX.get(store, "file.txt", if_none_match: cached_etag) do
        {:ok, data, meta} -> {:modified, data, meta}
        {:error, :not_modified} -> :use_cache
      end

      # Consistent read with ETag
      {:ok, meta} = ObjectStoreX.head(store, "config.json")
      case ObjectStoreX.get(store, "config.json", if_match: meta[:etag]) do
        {:ok, data, _meta} -> {:ok, data}
        {:error, :precondition_failed} -> :retry
      end

      # Range read
      {:ok, data, meta} = ObjectStoreX.get(store, "large.bin", range: {0, 1000})

      # Head-only (metadata without content)
      {:ok, _empty, meta} = ObjectStoreX.get(store, "file.txt", head: true)
  """
  @spec get(store(), path(), keyword()) ::
          {:ok, binary()} | {:ok, binary(), metadata()} | {:error, term()}
  def get(store, path, opts \\ [])

  def get(store, path, []) do
    case Native.get(store, path) do
      data when is_binary(data) -> {:ok, data}
      :not_found -> {:error, :not_found}
      error -> {:error, error}
    end
  rescue
    e -> {:error, Exception.message(e)}
  end

  def get(store, path, opts) when is_list(opts) do
    # Convert keyword options to GetOptions struct
    get_options = %ObjectStoreX.GetOptions{
      if_match: Keyword.get(opts, :if_match),
      if_none_match: Keyword.get(opts, :if_none_match),
      if_modified_since: convert_datetime_to_timestamp(Keyword.get(opts, :if_modified_since)),
      if_unmodified_since: convert_datetime_to_timestamp(Keyword.get(opts, :if_unmodified_since)),
      range: convert_range(Keyword.get(opts, :range)),
      version: Keyword.get(opts, :version),
      head: Keyword.get(opts, :head, false)
    }

    case Native.get_with_options(store, path, get_options) do
      {:ok, data, meta} when is_map(meta) ->
        # Convert charlist to binary if needed
        binary_data = if is_list(data), do: :erlang.list_to_binary(data), else: data
        {:ok, binary_data, meta}

      :not_found ->
        {:error, :not_found}

      :not_modified ->
        {:error, :not_modified}

      :precondition_failed ->
        {:error, :precondition_failed}

      error ->
        {:error, error}
    end
  rescue
    e -> {:error, Exception.message(e)}
  end

  # Convert DateTime to Unix timestamp, or pass through integer timestamps
  defp convert_datetime_to_timestamp(nil), do: nil
  defp convert_datetime_to_timestamp(%DateTime{} = dt), do: DateTime.to_unix(dt)
  defp convert_datetime_to_timestamp(ts) when is_integer(ts), do: ts

  # Convert range tuple or Range struct to Range struct
  defp convert_range(nil), do: nil
  defp convert_range({start, end_pos}), do: %ObjectStoreX.Range{start: start, end: end_pos}
  defp convert_range(%ObjectStoreX.Range{} = range), do: range

  @doc """
  Delete an object from storage.

  ## Examples

      :ok = ObjectStoreX.delete(store, "file.txt")
  """
  @spec delete(store(), path()) :: :ok | {:error, term()}
  def delete(store, path) do
    case Native.delete(store, path) do
      :ok -> :ok
      error -> {:error, error}
    end
  rescue
    e -> {:error, Exception.message(e)}
  end

  @doc """
  Get object metadata without downloading content.

  ## Examples

      {:ok, meta} = ObjectStoreX.head(store, "file.txt")
      # %{location: "file.txt", size: 1024, ...}
  """
  @spec head(store(), path()) :: {:ok, metadata()} | {:error, term()}
  def head(store, path) do
    case Native.head(store, path) do
      meta when is_map(meta) -> {:ok, meta}
      :not_found -> {:error, :not_found}
      error -> {:error, error}
    end
  rescue
    e -> {:error, Exception.message(e)}
  end

  @doc """
  Copy an object within storage (server-side).

  ## Examples

      :ok = ObjectStoreX.copy(store, "source.txt", "destination.txt")
  """
  @spec copy(store(), path(), path()) :: :ok | {:error, term()}
  def copy(store, from, to) do
    case Native.copy(store, from, to) do
      :ok -> :ok
      error -> {:error, error}
    end
  rescue
    e -> {:error, Exception.message(e)}
  end

  @doc """
  Rename an object (server-side move).

  ## Examples

      :ok = ObjectStoreX.rename(store, "old.txt", "new.txt")
  """
  @spec rename(store(), path(), path()) :: :ok | {:error, term()}
  def rename(store, from, to) do
    case Native.rename(store, from, to) do
      :ok -> :ok
      error -> {:error, error}
    end
  rescue
    e -> {:error, Exception.message(e)}
  end

  @doc """
  Copy an object only if the destination doesn't exist (atomic where supported).

  Provider support:
  - Azure: ✅ Native atomic copy_if_not_exists
  - GCS: ✅ Native atomic copy_if_not_exists
  - Local: ✅ Atomic via filesystem operations
  - Memory: ✅ Atomic via in-memory checks
  - S3: ❌ Not supported (returns `{:error, :not_supported}`)

  For S3, use a manual check-then-copy pattern:

      case ObjectStoreX.head(store, destination) do
        {:error, :not_found} ->
          ObjectStoreX.copy(store, source, destination)
        {:ok, _meta} ->
          {:error, :already_exists}
      end

  ## Examples

      # Atomic copy (Azure/GCS/Local/Memory)
      case ObjectStoreX.copy_if_not_exists(store, "source.txt", "backup.txt") do
        :ok -> :copied
        {:error, :already_exists} -> :destination_exists
        {:error, :not_supported} -> :use_manual_pattern
      end

      # Distributed lock backup
      case ObjectStoreX.copy_if_not_exists(store, "lock.txt", "lock-backup.txt") do
        :ok -> :backup_created
        {:error, :already_exists} -> :backup_already_exists
      end
  """
  @spec copy_if_not_exists(store(), path(), path()) :: :ok | {:error, term()}
  def copy_if_not_exists(store, from, to) do
    case Native.copy_if_not_exists(store, from, to) do
      :ok -> :ok
      :already_exists -> {:error, :already_exists}
      :not_supported -> {:error, :not_supported}
      :not_found -> {:error, :not_found}
      error -> {:error, error}
    end
  rescue
    e -> {:error, Exception.message(e)}
  end

  @doc """
  Rename an object only if the destination doesn't exist (atomic where supported).

  This is implemented as copy_if_not_exists followed by delete of the source.
  The operation is only atomic if the underlying provider supports atomic copy_if_not_exists.

  Provider support:
  - Azure: ✅ Atomic
  - GCS: ✅ Atomic
  - Local: ✅ Atomic
  - Memory: ✅ Atomic
  - S3: ❌ Not supported (returns `{:error, :not_supported}`)

  ## Examples

      # Atomic rename (Azure/GCS/Local/Memory)
      case ObjectStoreX.rename_if_not_exists(store, "old.txt", "new.txt") do
        :ok -> :renamed
        {:error, :already_exists} -> :destination_exists
        {:error, :not_supported} -> :use_manual_pattern
      end

      # Safe rename with collision detection
      case ObjectStoreX.rename_if_not_exists(store, "temp.txt", "final.txt") do
        :ok -> :moved
        {:error, :already_exists} -> :collision
        {:error, :not_found} -> :source_missing
      end
  """
  @spec rename_if_not_exists(store(), path(), path()) :: :ok | {:error, term()}
  def rename_if_not_exists(store, from, to) do
    case Native.rename_if_not_exists(store, from, to) do
      :ok -> :ok
      :already_exists -> {:error, :already_exists}
      :not_supported -> {:error, :not_supported}
      :not_found -> {:error, :not_found}
      error -> {:error, error}
    end
  rescue
    e -> {:error, Exception.message(e)}
  end

  @doc """
  Fetch multiple byte ranges from an object in a single operation.

  Useful for reading file headers, footers, and metadata without downloading
  the entire file (e.g., Parquet files, video files).

  ## Examples

      # Read header and footer
      {:ok, [header, footer]} = ObjectStoreX.get_ranges(store, "data.parquet", [
        {0, 1000},           # First 1000 bytes
        {9000, 10000}        # Bytes 9000-10000
      ])

      # Read specific sections
      {:ok, chunks} = ObjectStoreX.get_ranges(store, "video.mp4", [
        {0, 100},            # Header
        {500, 600},          # Metadata
        {1000, 2000}         # Preview section
      ])
  """
  @spec get_ranges(store(), path(), [{non_neg_integer(), non_neg_integer()}]) ::
          {:ok, [binary()]} | {:error, term()}
  def get_ranges(store, path, ranges) when is_list(ranges) do
    case Native.get_ranges(store, path, ranges) do
      binaries when is_list(binaries) -> {:ok, binaries}
      :not_found -> {:error, :not_found}
      error -> {:error, error}
    end
  rescue
    e -> {:error, Exception.message(e)}
  end

  @doc """
  Delete multiple objects in bulk with automatic batching.

  Provider-specific batching:
  - S3: Up to 1000 objects per request
  - Azure: Individual deletes (parallelized)
  - GCS: Batch API (up to 100 objects per request)

  ## Examples

      # Delete many objects
      paths = ["file1.txt", "file2.txt", "file3.txt"]
      {:ok, 3, []} = ObjectStoreX.delete_many(store, paths)

      # Handle partial failures
      result = ObjectStoreX.delete_many(store, paths)
      case result do
        {:ok, succeeded, failed} ->
          IO.puts("Deleted: \#{succeeded}, Failed: \#{length(failed)}")
        {:error, reason} ->
          IO.puts("Error: \#{reason}")
      end
  """
  @spec delete_many(store(), [path()]) :: {:ok, non_neg_integer(), list()} | {:error, term()}
  def delete_many(store, paths) when is_list(paths) do
    case Native.delete_many(store, paths) do
      {succeeded, failed} when is_integer(succeeded) and is_list(failed) ->
        {:ok, succeeded, failed}

      error ->
        {:error, error}
    end
  rescue
    e -> {:error, Exception.message(e)}
  end

  @doc """
  List objects with delimiter, returning objects and common prefixes separately.

  This simulates a directory structure by grouping objects by common prefixes.
  The delimiter is always "/" (forward slash).

  ## Examples

      {:ok, objects, prefixes} = ObjectStoreX.list_with_delimiter(store, prefix: "data/")

      # objects = [%{location: "data/file.txt", ...}]
      # prefixes = ["data/2024/", "data/2025/"]

      # List root level
      {:ok, objects, prefixes} = ObjectStoreX.list_with_delimiter(store)

  ## Options

  * `:prefix` - Optional prefix to filter objects (default: nil)

  ## Returns

  A tuple of `{:ok, objects, prefixes}` where:
  - `objects` is a list of metadata maps for objects at the current level
  - `prefixes` is a list of string prefixes (subdirectories)
  """
  @spec list_with_delimiter(store(), keyword()) ::
          {:ok, [metadata()], [String.t()]} | {:error, term()}
  def list_with_delimiter(store, opts \\ []) do
    prefix = Keyword.get(opts, :prefix)

    case Native.list_with_delimiter(store, prefix) do
      {objects, prefixes} when is_list(objects) and is_list(prefixes) ->
        {:ok, objects, prefixes}

      error ->
        {:error, error}
    end
  rescue
    e -> {:error, Exception.message(e)}
  end
end