phoenix_kit_emails

defmodule PhoenixKit.Modules.Emails.Archiver do
  @moduledoc """
  Archive and compress old email tracking data for optimal storage.

  Provides comprehensive data lifecycle management for email tracking:

  - **Body Compression** - Compress full email bodies after configurable time
  - **S3 Archival** - Move old logs to S3 cold storage  
  - **Sampling Optimization** - Apply sampling to reduce storage load
  - **Cleanup Integration** - Work with cleanup tasks for complete lifecycle
  - **Performance Optimization** - Batch operations for large datasets

  ## Storage Optimization Strategy

  1. **Recent Data** (0-7 days): Full storage with all fields
  2. **Medium Data** (7-30 days): Compress body_full, keep metadata
  3. **Old Data** (30-90 days): Archive to S3, keep local summary
  4. **Ancient Data** (90+ days): Delete after S3 confirmation

  ## Settings Integration

  All archival settings stored in phoenix_kit_settings:

  - `email_compress_body` - Days before compressing bodies (default: 30)
  - `email_archive_to_s3` - Enable S3 archival (default: false)
  - `email_s3_bucket` - S3 bucket name
  - `email_sampling_rate` - Percentage to fully log (default: 100)
  - `email_retention_days` - Total retention before deletion (default: 90)

  ## Usage Examples

      # Compress bodies older than 30 days
      {compressed_count, size_saved} = PhoenixKit.Modules.Emails.Archiver.compress_old_bodies(30)

      # Archive to S3 with automatic cleanup
      {:ok, archived_count} = PhoenixKit.Modules.Emails.Archiver.archive_to_s3(90, 
        bucket: "my-email-archive",
        prefix: "email-logs/2025/"
      )

      # Apply sampling to reduce future storage
      sampled_email = PhoenixKit.Modules.Emails.Archiver.apply_sampling_rate(email)

      # Get storage statistics
      stats = PhoenixKit.Modules.Emails.Archiver.get_storage_stats()
      # => %{total_logs: 50000, compressed: 15000, archived: 10000, size_mb: 2341}

  ## S3 Integration

  Supports multiple S3-compatible storage providers:
  - Amazon S3
  - DigitalOcean Spaces  
  - Google Cloud Storage
  - MinIO
  - Any S3-compatible service

  ## Compression Algorithm

  Uses gzip compression for email bodies with fallback strategies:

  1. **Gzip** - Primary compression for text content
  2. **Preview Only** - Keep only first 500 chars for very old data
  3. **Metadata Only** - Keep only delivery status and timestamps

  ## Batch Processing

  All operations are designed for efficiency:
  - Process in configurable batch sizes (default: 1000)
  - Progress tracking for long operations
  - Automatic retry on transient failures
  - Memory-efficient streaming for large datasets
  """

  require Logger
  alias PhoenixKit.Modules.Emails.{Event, Log}
  alias PhoenixKit.Settings
  alias PhoenixKit.Utils.Date, as: UtilsDate
  import Ecto.Query

  ## --- Body Compression ---

  @doc """
  Compress email bodies older than specified days.

  Returns `{compressed_count, size_saved_bytes}`.

  ## Options

  - `:batch_size` - Process in batches (default: 1000)
  - `:dry_run` - Show what would be compressed without doing it
  - `:preserve_errors` - Don't compress emails with errors/bounces

  ## Examples

      # Compress bodies older than 30 days
      {count, saved} = Archiver.compress_old_bodies(30)
      # => {1523, 45231040}

      # Dry run to see impact
      {count, estimated_saved} = Archiver.compress_old_bodies(30, dry_run: true)
  """
  def compress_old_bodies(days_old \\ nil, opts \\ []) do
    days_old = days_old || get_compress_days()
    batch_size = Keyword.get(opts, :batch_size, 1000)
    dry_run = Keyword.get(opts, :dry_run, false)
    preserve_errors = Keyword.get(opts, :preserve_errors, true)

    Logger.info("Starting body compression for emails older than #{days_old} days")

    cutoff_date = DateTime.add(UtilsDate.utc_now(), -days_old * 86_400)

    query = build_compression_query(cutoff_date, preserve_errors)

    if dry_run do
      {count, estimated_size} = estimate_compression_savings(query)
      Logger.info("Would compress #{count} email bodies, saving ~#{format_bytes(estimated_size)}")
      {count, estimated_size}
    else
      process_compression_batches(query, batch_size)
    end
  end

  @doc """
  Apply sampling rate to email for storage optimization.

  Returns modified email with reduced storage footprint for non-critical emails.

  ## Sampling Strategy

  - **Always Full**: Error emails, bounces, complaints  
  - **Always Full**: Transactional emails (password resets, etc.)
  - **Sampling Applied**: Marketing emails, newsletters
  - **Metadata Only**: Bulk emails when over limit

  ## Examples

      # Apply system sampling rate
      email = Archiver.apply_sampling_rate(original_email)

      # Force specific sampling
      email = Archiver.apply_sampling_rate(original_email, force_rate: 50)
  """
  def apply_sampling_rate(email_attrs, opts \\ []) do
    sampling_rate = Keyword.get(opts, :force_rate) || get_sampling_rate()

    # Always store critical emails fully
    if critical_email?(email_attrs) do
      email_attrs
    else
      random_value = :rand.uniform(100)

      if random_value <= sampling_rate do
        # Store fully
        email_attrs
      else
        # Store with reduced data
        apply_reduced_storage(email_attrs)
      end
    end
  end

  ## --- S3 Archival ---

  @doc """
  Archive old emails to S3 storage.

  Returns `{:ok, archived_count}` on success or `{:error, reason}` on failure.

  ## Options

  - `:bucket` - S3 bucket name (required)
  - `:prefix` - S3 object key prefix
  - `:batch_size` - Process in batches (default: 500) 
  - `:format` - Archive format: :json (default), :csv, :parquet
  - `:delete_after_archive` - Delete from DB after successful archive
  - `:include_events` - Include email events in archive

  ## Examples

      # Basic S3 archival
      {:ok, count} = Archiver.archive_to_s3(90,
        bucket: "email-archive",
        prefix: "logs/2025/"
      )

      # Archive with events and cleanup
      {:ok, count} = Archiver.archive_to_s3(90,
        bucket: "email-archive", 
        include_events: true,
        delete_after_archive: true
      )
  """
  def archive_to_s3(days_old, opts \\ []) do
    if s3_archival_enabled?() do
      bucket = Keyword.get(opts, :bucket) || get_s3_bucket()
      prefix = Keyword.get(opts, :prefix, "email-logs/")
      batch_size = Keyword.get(opts, :batch_size, 500)
      format = Keyword.get(opts, :format, :json)
      delete_after = Keyword.get(opts, :delete_after_archive, false)
      include_events = Keyword.get(opts, :include_events, true)

      if bucket do
        do_s3_archival(days_old, bucket, prefix, batch_size, format, delete_after, include_events)
      else
        {:error, :no_bucket_configured}
      end
    else
      {:error, :s3_not_configured}
    end
  end

  defp do_s3_archival(days_old, bucket, prefix, batch_size, format, delete_after, include_events) do
    Logger.info("Starting S3 archival for emails older than #{days_old} days")

    cutoff_date = DateTime.add(UtilsDate.utc_now(), -days_old * 86_400)
    query = build_archival_query(cutoff_date)

    case process_s3_archival(
           query,
           bucket,
           prefix,
           batch_size,
           format,
           include_events,
           delete_after
         ) do
      {:ok, archived_count} ->
        Logger.info("Successfully archived #{archived_count} emails to S3")
        {:ok, archived_count}

      {:error, reason} ->
        Logger.error("S3 archival failed: #{inspect(reason)}")
        {:error, reason}
    end
  end

  ## --- Storage Statistics ---

  @doc """
  Get comprehensive storage statistics.

  ## Examples

      iex> Archiver.get_storage_stats()
      %{
        total_logs: 125000,
        total_events: 450000,
        compressed_bodies: 45000,
        archived_logs: 15000,
        storage_size_mb: 2341,
        oldest_log: ~U[2024-01-15 10:30:00Z],
        compression_ratio: 0.65,
        s3_archived_size_mb: 890
      }
  """
  def get_storage_stats do
    %{
      total_logs: count_total_logs(),
      total_events: count_total_events(),
      compressed_bodies: count_compressed_bodies(),
      archived_logs: count_archived_logs(),
      storage_size_mb: calculate_storage_size_mb(),
      oldest_log: get_oldest_log_date(),
      compression_ratio: calculate_compression_ratio(),
      s3_archived_size_mb: get_s3_archived_size()
    }
  end

  @doc """
  Get detailed storage breakdown by time periods.

  ## Examples

      iex> Archiver.get_storage_breakdown()
      %{
        last_7_days: %{logs: 5000, size_mb: 145, compressed: false},
        last_30_days: %{logs: 15000, size_mb: 420, compressed: 8000},
        last_90_days: %{logs: 35000, size_mb: 980, compressed: 25000},
        older: %{logs: 70000, size_mb: 1200, archived: 45000}
      }
  """
  def get_storage_breakdown do
    now = UtilsDate.utc_now()

    %{
      last_7_days: get_period_stats(DateTime.add(now, -7 * 86_400), now),
      last_30_days: get_period_stats(DateTime.add(now, -30 * 86_400), now),
      last_90_days: get_period_stats(DateTime.add(now, -90 * 86_400), now),
      older: get_period_stats(~U[1970-01-01 00:00:00Z], DateTime.add(now, -90 * 86_400))
    }
  end

  ## --- Configuration Helpers ---

  defp get_compress_days do
    Settings.get_integer_setting("email_compress_body", 30)
  end

  defp get_sampling_rate do
    Settings.get_integer_setting("email_sampling_rate", 100)
  end

  defp s3_archival_enabled? do
    Settings.get_boolean_setting("email_archive_to_s3", false)
  end

  defp get_s3_bucket do
    Settings.get_setting("email_s3_bucket")
  end

  ## --- Query Builders ---

  defp build_compression_query(cutoff_date, preserve_errors) do
    query =
      from(l in Log,
        where: l.sent_at < ^cutoff_date,
        where: not is_nil(l.body_full),
        where: l.body_full != ""
      )

    if preserve_errors do
      query
      |> where([l], l.status not in ["bounced", "failed", "complained"])
    else
      query
    end
  end

  defp build_archival_query(cutoff_date) do
    from(l in Log,
      where: l.sent_at < ^cutoff_date,
      order_by: [asc: l.sent_at]
    )
  end

  ## --- Compression Implementation ---

  defp estimate_compression_savings(query) do
    stats_query =
      from(l in query,
        select: {count(l.uuid), sum(fragment("LENGTH(?)", l.body_full))}
      )

    case repo().one(stats_query) do
      {count, total_size} when not is_nil(total_size) ->
        # Estimate 60% compression ratio for email bodies
        estimated_savings = trunc(total_size * 0.6)
        {count || 0, estimated_savings}

      _ ->
        {0, 0}
    end
  end

  defp process_compression_batches(query, batch_size) do
    query
    |> stream_in_batches(batch_size, fn batch ->
      {batch_compressed, batch_saved} = compress_batch(batch)
      {batch_compressed, batch_saved}
    end)
    |> Enum.reduce({0, 0}, fn {count, saved}, {total_count, total_saved} ->
      {total_count + count, total_saved + saved}
    end)
  rescue
    # Mirror process_s3_archival's resilience: a raised exception inside the
    # streaming transaction (the whole run rolls back atomically) must not crash
    # the background compaction job. Report zero progress for this run.
    error ->
      Logger.error("Email archiver: body compression run failed: #{Exception.message(error)}")
      {0, 0}
  end

  defp compress_batch(email_logs) do
    Enum.reduce(email_logs, {0, 0}, fn log, {count, saved} ->
      case compress_email_body(log) do
        {:ok, size_saved} -> {count + 1, saved + size_saved}
        {:error, _} -> {count, saved}
      end
    end)
  end

  defp compress_email_body(%Log{} = log) do
    if log.body_full && String.length(log.body_full) > 100 do
      original_size = byte_size(log.body_full)

      # Compress with gzip
      compressed_data = :zlib.gzip(log.body_full)
      compressed_size = byte_size(compressed_data)

      # Only compress if we save significant space
      if compressed_size < original_size * 0.8 do
        case repo().update(
               Log.changeset(log, %{
                 body_full: Base.encode64(compressed_data)
               })
             ) do
          {:ok, _} -> {:ok, original_size - compressed_size}
          {:error, changeset} -> {:error, changeset}
        end
      else
        # Compression not worth it, just keep preview
        case repo().update(
               Log.changeset(log, %{
                 body_full: nil,
                 body_preview: String.slice(log.body_full, 0, 500)
               })
             ) do
          {:ok, _} -> {:ok, original_size}
          {:error, changeset} -> {:error, changeset}
        end
      end
    else
      {:ok, 0}
    end
  end

  ## --- S3 Implementation ---

  defp process_s3_archival(
         query,
         bucket,
         prefix,
         batch_size,
         format,
         include_events,
         delete_after
       ) do
    _archived_count = 0

    try do
      query
      |> stream_in_batches(batch_size, fn batch ->
        archive_batch_to_s3(batch, bucket, prefix, format, include_events, delete_after)
      end)
      |> Enum.reduce(0, fn batch_count, total -> total + batch_count end)
      |> then(fn count -> {:ok, count} end)
    rescue
      error -> {:error, Exception.message(error)}
    end
  end

  @dialyzer {:nowarn_function, archive_batch_to_s3: 6}
  defp archive_batch_to_s3(logs, bucket, prefix, format, include_events, delete_after) do
    timestamp = UtilsDate.utc_now() |> DateTime.to_iso8601()
    batch_id = :crypto.strong_rand_bytes(8) |> Base.encode16(case: :lower)

    # Prepare data
    archive_data = prepare_archive_data(logs, format, include_events)

    # Generate S3 key
    s3_key = "#{prefix}#{timestamp}/batch-#{batch_id}.#{format}"

    # Upload to S3
    case upload_to_s3(bucket, s3_key, archive_data) do
      {:ok, _message} ->
        if delete_after do
          delete_archived_logs(logs)
        end

        length(logs)

      {:error, reason} ->
        Logger.error("Failed to archive batch to S3: #{inspect(reason)}")
        0
    end
  end

  defp prepare_archive_data(logs, :json, include_events) do
    archive_logs =
      if include_events do
        Enum.map(logs, fn log ->
          events = repo().all(from(e in Event, where: e.email_log_uuid == ^log.uuid))
          Map.put(log, :events, events)
        end)
      else
        logs
      end

    JSON.encode!(%{
      exported_at: UtilsDate.utc_now(),
      total_records: length(logs),
      logs: archive_logs
    })
  end

  defp prepare_archive_data(logs, :csv, _include_events) do
    # CSV format implementation
    header = "uuid,message_id,to,from,subject,status,sent_at,delivered_at\n"

    rows =
      logs
      |> Enum.map_join("\n", fn log ->
        [
          log.uuid,
          log.message_id,
          log.to,
          log.from,
          log.subject,
          log.status,
          log.sent_at,
          log.delivered_at
        ]
        |> Enum.map_join(",", &csv_escape/1)
      end)

    header <> rows
  end

  @spec upload_to_s3(String.t(), String.t(), binary()) :: {:ok, String.t()} | {:error, String.t()}
  defp upload_to_s3(bucket, key, data) do
    # Upload compressed data to S3 with proper error handling
    case ExAws.S3.put_object(bucket, key, data,
           content_type: "application/gzip",
           content_encoding: "gzip",
           metadata: %{
             "archived-by" => "phoenix_kit",
             "archived-at" => DateTime.to_iso8601(UtilsDate.utc_now())
           }
         )
         |> ExAws.request() do
      {:ok, _result} ->
        Logger.info("Successfully uploaded archive to S3: s3://#{bucket}/#{key}")
        {:ok, "Successfully archived to S3: #{key}"}

      {:error, {:http_error, 404, _}} ->
        Logger.error("S3 bucket not found: #{bucket}")
        {:error, "S3 bucket not found. Please ensure bucket '#{bucket}' exists."}

      {:error, {:http_error, 403, _}} ->
        Logger.error("S3 access denied for bucket: #{bucket}")
        {:error, "Access denied to S3 bucket. Check IAM permissions."}

      {:error, reason} ->
        Logger.error("Failed to upload to S3: #{inspect(reason)}")
        {:error, "S3 upload failed: #{inspect(reason)}"}
    end
  rescue
    error ->
      Logger.error("S3 upload exception: #{inspect(error)}")
      {:error, "S3 upload exception: #{Exception.message(error)}"}
  end

  @dialyzer {:nowarn_function, delete_archived_logs: 1}
  defp delete_archived_logs(logs) do
    log_uuids = Enum.map(logs, & &1.uuid)

    # Delete events first (foreign key constraint)
    from(e in Event, where: e.email_log_uuid in ^log_uuids)
    |> repo().delete_all()

    # Delete logs
    from(l in Log, where: l.uuid in ^log_uuids)
    |> repo().delete_all()
  end

  ## --- Sampling Implementation ---

  defp critical_email?(email_attrs) do
    # Check if this is a critical email that should always be stored fully
    cond do
      email_attrs[:status] in ["bounced", "failed", "complained"] -> true
      String.contains?(email_attrs[:subject] || "", ["password", "reset", "verify"]) -> true
      email_attrs[:template_name] in ["password_reset", "email_confirmation"] -> true
      true -> false
    end
  end

  defp apply_reduced_storage(email_attrs) do
    # Store only essential fields for sampled emails
    Map.take(email_attrs, [
      :message_id,
      :to,
      :from,
      :subject,
      :status,
      :sent_at,
      :delivered_at,
      :provider,
      :campaign_id,
      :template_name
    ])
  end

  ## --- Statistics Implementation ---

  defp count_total_logs do
    repo().one(from(l in Log, select: count(l.uuid))) || 0
  end

  defp count_total_events do
    repo().one(from(e in Event, select: count(e.uuid))) || 0
  end

  defp count_compressed_bodies do
    # Count emails with base64-encoded compressed bodies
    repo().one(
      from(l in Log,
        where: fragment("? LIKE 'H4sI%'", l.body_full),
        select: count(l.uuid)
      )
    ) || 0
  end

  defp count_archived_logs do
    # This would count logs marked as archived
    # Simplified for now
    0
  end

  defp calculate_storage_size_mb do
    # Estimate storage size based on average email size
    total_logs = count_total_logs()
    total_events = count_total_events()

    # Rough estimates: 2KB per log, 0.5KB per event
    estimated_bytes = total_logs * 2048 + total_events * 512
    Float.round(estimated_bytes / 1024 / 1024, 1)
  end

  defp get_oldest_log_date do
    repo().one(from(l in Log, select: min(l.sent_at)))
  end

  defp calculate_compression_ratio do
    compressed_count = count_compressed_bodies()
    total_count = count_total_logs()

    if total_count > 0 do
      Float.round(compressed_count / total_count, 2)
    else
      0.0
    end
  end

  defp get_s3_archived_size do
    # This would query S3 for archived data size
    # Simplified for now
    0.0
  end

  defp get_period_stats(start_time, end_time) do
    query = from(l in Log, where: l.sent_at >= ^start_time and l.sent_at < ^end_time)

    count = repo().one(from(l in query, select: count(l.uuid))) || 0

    compressed =
      repo().one(
        from(l in query,
          where: fragment("? LIKE 'H4sI%'", l.body_full),
          select: count(l.uuid)
        )
      ) || 0

    %{
      logs: count,
      compressed: compressed,
      # Estimated
      size_mb: Float.round(count * 2.048, 1)
    }
  end

  ## --- Utility Helpers ---

  # Truly streams the query in chunks of `batch_size`, applying `mapper_func` to
  # each chunk. Uses Repo.stream (which requires an enclosing transaction) so the
  # full result set — including large body_full blobs — is never loaded into
  # memory at once. Returns the list of per-batch mapper results.
  defp stream_in_batches(query, batch_size, mapper_func) do
    {:ok, results} =
      repo().transaction(
        fn ->
          query
          |> repo().stream(max_rows: batch_size)
          |> Stream.chunk_every(batch_size)
          |> Stream.map(mapper_func)
          |> Enum.to_list()
        end,
        timeout: :infinity
      )

    results
  end

  defp format_bytes(bytes) when bytes < 1024, do: "#{bytes}B"
  defp format_bytes(bytes) when bytes < 1024 * 1024, do: "#{Float.round(bytes / 1024, 1)}KB"

  defp format_bytes(bytes) when bytes < 1024 * 1024 * 1024,
    do: "#{Float.round(bytes / 1024 / 1024, 1)}MB"

  defp format_bytes(bytes), do: "#{Float.round(bytes / 1024 / 1024 / 1024, 1)}GB"

  defp csv_escape(nil), do: ""

  defp csv_escape(value) when is_binary(value) do
    if String.contains?(value, [",", "\"", "\n"]) do
      "\"#{String.replace(value, "\"", "\"\"")}\""
    else
      value
    end
  end

  defp csv_escape(value), do: to_string(value)

  # Gets the configured repository for database operations
  defp repo do
    PhoenixKit.RepoHelper.repo()
  end
end