adaptive_backfill

defmodule Examples.RealWorldPostgresExample do
  @moduledoc """
  Real-world examples using PostgreSQL with Ecto.
  
  These examples demonstrate practical use cases with actual database
  operations and the built-in PostgreSQL health checkers.
  """

  use AdaptiveBackfill
  alias AdaptiveBackfill.DefaultPgHealthCheckers

  # Assume you have a Repo module
  # alias MyApp.Repo

  # Example 1: Backfill missing data in existing records
  batch_operation :backfill_user_metadata, initial_state: 0 do
    mode :sync

    # Use built-in PostgreSQL health checkers
    health_checks DefaultPgHealthCheckers.pg_health_checks(MyApp.Repo)

    handle_batch fn offset ->
      IO.puts("\n--- Backfilling users at offset #{offset} ---")

      # Query users missing metadata using Ecto
      users =
        MyApp.Repo.all(
          from u in "users",
            where: is_nil(u.metadata),
            offset: ^offset,
            limit: 100,
            select: %{id: u.id, email: u.email}
        )

      if Enum.empty?(users) do
        IO.puts("No more users to backfill")
        :done
      else
        # Backfill metadata for each user
        Enum.each(users, fn user ->
          metadata = generate_metadata_for_user(user)

          MyApp.Repo.query!(
            "UPDATE users SET metadata = $1, updated_at = NOW() WHERE id = $2",
            [metadata, user.id]
          )
        end)

        IO.puts("  ✓ Backfilled #{length(users)} users")
        {:ok, offset + 100}
      end
    end

    on_complete fn result ->
      case result do
        :done -> IO.puts("\n✓ User metadata backfill completed!")
        {:halt, reason} -> IO.puts("\n✗ Backfill halted: #{inspect(reason)}")
      end
    end
  end

  # Example 2: Data migration between tables
  batch_operation :migrate_orders_to_new_schema, initial_state: 0 do
    mode :async
    health_checks [
      &DefaultPgHealthCheckers.long_waiting_queries/1,
      &DefaultPgHealthCheckers.hot_io_tables/1
    ]

    handle_batch fn offset ->
      IO.puts("\n--- Migrating orders at offset #{offset} ---")

      # Fetch old orders
      orders =
        MyApp.Repo.all(
          from o in "old_orders",
            where: o.migrated == false,
            offset: ^offset,
            limit: 50,
            select: %{
              id: o.id,
              user_id: o.user_id,
              total: o.total,
              items: o.items,
              created_at: o.created_at
            }
        )

      if Enum.empty?(orders) do
        :done
      else
        # Insert into new schema with transaction
        MyApp.Repo.transaction(fn ->
          Enum.each(orders, fn order ->
            # Insert into new orders table
            MyApp.Repo.query!(
              """
              INSERT INTO orders (user_id, total_amount, status, created_at)
              VALUES ($1, $2, 'completed', $3)
              RETURNING id
              """,
              [order.user_id, order.total, order.created_at]
            )
            |> then(fn %{rows: [[new_order_id]]} ->
              # Insert order items
              insert_order_items(new_order_id, order.items)

              # Mark old order as migrated
              MyApp.Repo.query!(
                "UPDATE old_orders SET migrated = true WHERE id = $1",
                [order.id]
              )
            end)
          end)
        end)

        IO.puts("  ✓ Migrated #{length(orders)} orders")
        {:ok, offset + 50}
      end
    end

    on_complete fn result ->
      case result do
        :done -> IO.puts("\n✓ Order migration completed!")
        {:halt, reason} -> IO.puts("\n✗ Migration halted: #{inspect(reason)}")
      end
    end
  end

  # Example 3: Cleanup old data with safety checks
  batch_operation :cleanup_old_logs, initial_state: 0 do
    mode :sync
    health_checks [
      &DefaultPgHealthCheckers.temp_file_usage/1,
      &check_table_bloat/0
    ]

    handle_batch fn offset ->
      cutoff_date = DateTime.add(DateTime.utc_now(), -90, :day)
      IO.puts("\n--- Cleaning logs older than #{cutoff_date} ---")

      # Delete in batches to avoid long locks
      result =
        MyApp.Repo.query!(
          """
          DELETE FROM logs
          WHERE id IN (
            SELECT id FROM logs
            WHERE created_at < $1
            ORDER BY id
            LIMIT 1000
            OFFSET $2
          )
          """,
          [cutoff_date, offset]
        )

      deleted_count = result.num_rows

      if deleted_count == 0 do
        IO.puts("No more logs to delete")
        :done
      else
        IO.puts("  ✓ Deleted #{deleted_count} log entries")
        {:ok, offset + 1000}
      end
    end

    on_complete fn result ->
      case result do
        :done ->
          IO.puts("\n✓ Log cleanup completed!")
          IO.puts("Consider running VACUUM ANALYZE on the logs table")

        {:halt, reason} ->
          IO.puts("\n✗ Cleanup halted: #{inspect(reason)}")
      end
    end
  end

  # Helper functions
  defp generate_metadata_for_user(user) do
    Jason.encode!(%{
      email_domain: String.split(user.email, "@") |> List.last(),
      backfilled_at: DateTime.utc_now(),
      version: "1.0"
    })
  end

  defp insert_order_items(order_id, items_json) do
    items = Jason.decode!(items_json)

    Enum.each(items, fn item ->
      MyApp.Repo.query!(
        """
        INSERT INTO order_items (order_id, product_id, quantity, price)
        VALUES ($1, $2, $3, $4)
        """,
        [order_id, item["product_id"], item["quantity"], item["price"]]
      )
    end)
  end

  defp check_table_bloat do
    # Custom health check for table bloat
    result =
      MyApp.Repo.query!("""
        SELECT
          schemaname,
          tablename,
          pg_size_pretty(pg_total_relation_size(schemaname||'.'||tablename)) AS size
        FROM pg_tables
        WHERE schemaname = 'public'
        ORDER BY pg_total_relation_size(schemaname||'.'||tablename) DESC
        LIMIT 1
      """)

    case result.rows do
      [[_schema, table, size]] ->
        IO.puts("  ✓ Largest table: #{table} (#{size})")
        :ok

      _ ->
        :ok
    end
  end
end

# Run the examples
# Examples.RealWorldPostgresExample.backfill_user_metadata()
# Examples.RealWorldPostgresExample.migrate_orders_to_new_schema()
# Examples.RealWorldPostgresExample.cleanup_old_logs()