defmodule Mix.Tasks.Wnh.DbScaleBenchmark do
  use Mix.Task

  alias WhoNeedHelp.Repo

  @shortdoc "Measures keyset pagination plans on an isolated large data set"

  @moduledoc """
  Seeds and measures the isolated database used by `scripts/db-scale-benchmark.sh`.

  The task records PostgreSQL `EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON)` output.
  It intentionally does not enforce an execution-time threshold because timings
  depend on the host. The `after` phase does enforce query correctness and that
  each measured cursor query uses its purpose-built index.
  """

  @default_rows 50_000

  @impl Mix.Task
  def run(arguments) do
    Mix.Task.run("app.start")

    {options, _rest, invalid} =
      OptionParser.parse(arguments,
        strict: [phase: :string, rows: :integer, output: :string]
      )

    if invalid != [], do: Mix.raise("invalid options: #{inspect(invalid)}")

    phase = Keyword.get(options, :phase) || Mix.raise("--phase before|after is required")
    rows = Keyword.get(options, :rows, @default_rows)
    output = Keyword.get(options, :output) || Mix.raise("--output PATH is required")

    unless phase in ["before", "after"], do: Mix.raise("--phase must be before or after")
    unless rows >= 3, do: Mix.raise("--rows must be at least 3 for relational fixtures")

    File.mkdir_p!(output)

    if phase == "before" do
      seed!(rows)
    else
      assert_seed_size!(rows)
    end

    analyze!()
    plans = collect_plans!(phase, output)
    assert_keyset_pages!()

    summary = %{
      phase: phase,
      configured_rows_per_large_table: rows,
      captured_at: DateTime.utc_now() |> DateTime.truncate(:second) |> DateTime.to_iso8601(),
      postgres_version: scalar!("SHOW server_version"),
      table_counts: table_counts(),
      database_bytes: scalar!("SELECT pg_database_size(current_database())"),
      index_bytes: index_sizes(),
      plans: plans
    }

    summary_path = Path.join(output, "#{phase}-summary.json")
    File.write!(summary_path, Jason.encode_to_iodata!(summary, pretty: true))

    if phase == "after" do
      assert_expected_indexes_exist!()

      if rows == @default_rows do
        assert_expected_indexes_used!(plans)
      end

      write_comparison!(output, summary)
    end

    Mix.shell().info("database scale #{phase} phase written to #{summary_path}")
  end

  defp seed!(rows) do
    Repo.transaction(
      fn ->
        Enum.each(seed_statements(rows), &query!/1)
      end,
      timeout: :infinity
    )
  end

  defp seed_statements(rows) do
    [
      """
      INSERT INTO users (
        id, email, display_name, confirmed_at, accepted_terms_at, inserted_at, updated_at
      )
      SELECT
        md5('user-' || value)::uuid,
        'db-scale-' || value || '@example.invalid',
        'DB scale user ' || value,
        date_trunc('second', now()),
        date_trunc('second', now()),
        date_trunc('second', now()) - value * interval '1 second',
        date_trunc('second', now()) - value * interval '1 second'
      FROM generate_series(1, #{rows}) AS value
      """,
      """
      INSERT INTO categories (
        id, slug, names, descriptions, mode, inserted_at, updated_at
      )
      VALUES (
        md5('db-scale-category')::uuid,
        'db-scale-help',
        '{"en":"DB scale help"}'::jsonb,
        '{"en":"Synthetic isolated benchmark category"}'::jsonb,
        'help',
        date_trunc('second', now()),
        date_trunc('second', now())
      )
      """,
      """
      INSERT INTO help_requests (
        id, title, description, location_label, location, status, urgency,
        location_visibility, expires_at, hidden_at, requester_id, category_id,
        inserted_at, updated_at
      )
      SELECT
        md5('request-' || value)::uuid,
        'DB scale request ' || value,
        'Synthetic isolated database scale request ' || value,
        'Kyiv benchmark point',
        ST_SetSRID(ST_MakePoint(30.5 + (value % 100) / 10000.0, 50.4), 4326),
        CASE WHEN value % 4 = 0 THEN 'open' ELSE 'completed' END,
        'now',
        'approximate_public',
        date_trunc('second', now()) + (value % 30 + 1) * interval '1 day',
        CASE WHEN value % 20 = 0 THEN date_trunc('second', now()) END,
        md5('user-1')::uuid,
        md5('db-scale-category')::uuid,
        date_trunc('second', now()) - value * interval '1 second',
        date_trunc('second', now()) - value * interval '1 second'
      FROM generate_series(1, #{rows}) AS value
      """,
      """
      INSERT INTO help_assignments (
        id, status, handover_code_hash, handover_verified_at,
        proximity_observed_at, helper_movement_observed_at, accepted_at,
        completed_at, request_id, helper_id, inserted_at, updated_at
      )
      SELECT
        md5('assignment-' || value)::uuid,
        'completed',
        decode(md5('handover-' || value), 'hex'),
        CASE WHEN value % 2 = 0 THEN date_trunc('second', now()) END,
        CASE WHEN value % 3 = 0 THEN date_trunc('second', now()) END,
        CASE WHEN value % 4 = 0 THEN date_trunc('second', now()) END,
        date_trunc('second', now()) - value * interval '1 second',
        date_trunc('second', now()),
        md5('request-' || value)::uuid,
        md5('user-2')::uuid,
        date_trunc('second', now()) - value * interval '1 second',
        date_trunc('second', now())
      FROM generate_series(1, #{rows}) AS value
      """,
      """
      INSERT INTO messages (
        id, body, assignment_id, sender_id, inserted_at, updated_at
      )
      SELECT
        md5('message-' || value)::uuid,
        'Synthetic isolated message ' || value,
        md5('assignment-1')::uuid,
        md5('user-' || CASE WHEN value % 2 = 0 THEN 1 ELSE 2 END)::uuid,
        date_trunc('second', now()) - value * interval '1 second',
        date_trunc('second', now()) - value * interval '1 second'
      FROM generate_series(1, #{rows}) AS value
      """,
      """
      INSERT INTO reviews (
        id, rating, comment, revealed_at, assignment_id, reviewer_id, reviewee_id,
        inserted_at, updated_at
      )
      SELECT
        md5('review-' || value)::uuid,
        value % 5 + 1,
        'Synthetic isolated review ' || value,
        date_trunc('second', now()),
        md5('assignment-' || value)::uuid,
        md5('user-1')::uuid,
        md5('user-2')::uuid,
        date_trunc('second', now()) - value * interval '1 second',
        date_trunc('second', now()) - value * interval '1 second'
      FROM generate_series(1, #{rows}) AS value
      """,
      """
      INSERT INTO reports (
        id, reason, details, status, reporter_id, request_id, inserted_at, updated_at
      )
      SELECT
        md5('report-' || value)::uuid,
        'safety',
        'Synthetic isolated report ' || value,
        CASE
          WHEN value % 4 = 0 THEN 'open'
          WHEN value % 4 = 1 THEN 'reviewing'
          WHEN value % 4 = 2 THEN 'resolved'
          ELSE 'dismissed'
        END,
        md5('user-2')::uuid,
        md5('request-' || value)::uuid,
        date_trunc('second', now()) - value * interval '1 second',
        date_trunc('second', now()) - value * interval '1 second'
      FROM generate_series(1, #{rows}) AS value
      """,
      """
      INSERT INTO blocks (id, blocker_id, blocked_id, inserted_at)
      SELECT
        md5('block-' || value)::uuid,
        md5('user-2')::uuid,
        md5('user-' || value)::uuid,
        date_trunc('second', now()) - value * interval '1 second'
      FROM generate_series(3, #{rows}) AS value
      """,
      """
      INSERT INTO abuse_signals (
        id, kind, status, metadata, subject_id, inserted_at, updated_at
      )
      SELECT
        md5('signal-' || value)::uuid,
        'db_scale_' || value,
        CASE
          WHEN value % 3 = 0 THEN 'open'
          WHEN value % 3 = 1 THEN 'reviewed'
          ELSE 'dismissed'
        END,
        '{}'::jsonb,
        md5('user-' || value)::uuid,
        date_trunc('second', now()) - value * interval '1 second',
        date_trunc('second', now()) - value * interval '1 second'
      FROM generate_series(1, #{rows}) AS value
      """,
      """
      INSERT INTO category_proposals (
        id, proposed_name, reason, status, proposer_id, mode, inserted_at, updated_at
      )
      SELECT
        md5('proposal-' || value)::uuid,
        'DB scale proposal ' || value,
        'Synthetic isolated category proposal ' || value,
        CASE
          WHEN value % 4 = 0 THEN 'open'
          WHEN value % 4 = 1 THEN 'approved'
          WHEN value % 4 = 2 THEN 'rejected'
          ELSE 'merged'
        END,
        md5('user-1')::uuid,
        'help',
        date_trunc('second', now()) - value * interval '1 second',
        date_trunc('second', now()) - value * interval '1 second'
      FROM generate_series(1, #{rows}) AS value
      """,
      """
      INSERT INTO activities (
        id, title, description, location_label, location, status, starts_at,
        join_deadline, capacity, hidden_at, creator_id, category_id, inserted_at, updated_at
      )
      SELECT
        md5('activity-' || value)::uuid,
        'DB scale activity ' || value,
        'Synthetic isolated activity ' || value,
        'Kyiv benchmark point',
        ST_SetSRID(ST_MakePoint(30.5, 50.4 + (value % 100) / 10000.0), 4326),
        CASE
          WHEN value % 3 = 0 THEN 'open'
          WHEN value % 3 = 1 THEN 'completed'
          ELSE 'cancelled'
        END,
        date_trunc('second', now()) + (value + 1000) * interval '1 second',
        date_trunc('second', now()) + (value + 700) * interval '1 second',
        10,
        CASE WHEN value % 30 = 0 THEN date_trunc('second', now()) END,
        md5('user-1')::uuid,
        md5('db-scale-category')::uuid,
        date_trunc('second', now()) - value * interval '1 second',
        date_trunc('second', now()) - value * interval '1 second'
      FROM generate_series(1, #{rows}) AS value
      """,
      """
      INSERT INTO activity_participants (
        id, role, status, reviewed_at, activity_id, user_id, inserted_at, updated_at
      )
      SELECT
        md5('participant-' || value)::uuid,
        'participant',
        'approved',
        date_trunc('second', now()),
        md5('activity-' || value)::uuid,
        md5('user-' || (value % 100 + 3))::uuid,
        date_trunc('second', now()) - value * interval '1 second',
        date_trunc('second', now()) - value * interval '1 second'
      FROM generate_series(1, #{rows}) AS value
      UNION ALL
      SELECT
        md5('participant-user-2-' || value)::uuid,
        'participant',
        'approved',
        date_trunc('second', now()),
        md5('activity-' || value)::uuid,
        md5('user-2')::uuid,
        date_trunc('second', now()) - value * interval '1 second',
        date_trunc('second', now()) - value * interval '1 second'
      FROM generate_series(100, #{rows}, 100) AS value
      """,
      """
      INSERT INTO activity_messages (
        id, body, activity_id, sender_id, inserted_at, updated_at
      )
      SELECT
        md5('activity-message-' || value)::uuid,
        'Synthetic isolated activity message ' || value,
        md5('activity-1')::uuid,
        md5('user-1')::uuid,
        date_trunc('second', now()) - value * interval '1 second',
        date_trunc('second', now()) - value * interval '1 second'
      FROM generate_series(1, #{rows}) AS value
      """
    ]
  end

  defp analyze! do
    query!("""
    ANALYZE users, help_requests, help_assignments, messages, reviews, reports,
      blocks, abuse_signals, category_proposals, activities,
      activity_participants, activity_messages
    """)
  end

  defp collect_plans!(phase, output) do
    plan_queries()
    |> Map.new(fn {name, query} ->
      plan = explain!(query)

      File.write!(
        Path.join(output, "#{phase}-#{name}.json"),
        Jason.encode_to_iodata!(plan, pretty: true)
      )

      {name, summarize_plan(plan)}
    end)
  end

  defp plan_queries do
    viewer = "md5('user-1')::uuid"

    %{
      "help_discovery" => """
      SELECT id FROM help_requests
      WHERE status = 'open' AND hidden_at IS NULL AND expires_at > now()
        AND requester_id NOT IN (
          SELECT blocked_id FROM blocks WHERE blocker_id = #{viewer}
        )
        AND requester_id NOT IN (
          SELECT blocker_id FROM blocks WHERE blocked_id = #{viewer}
        )
      ORDER BY expires_at ASC, id ASC LIMIT 25
      """,
      "help_requester" => """
      SELECT id FROM help_requests
      WHERE requester_id = md5('user-1')::uuid
      ORDER BY inserted_at DESC, id DESC LIMIT 25
      """,
      "activity_discovery" => """
      SELECT id FROM activities
      WHERE status = 'open' AND hidden_at IS NULL
        AND starts_at > now() AND join_deadline > now()
      ORDER BY starts_at ASC, id ASC LIMIT 25
      """,
      "activity_participant" => """
      SELECT activities.id
      FROM activities
      INNER JOIN activity_participants
        ON activity_participants.activity_id = activities.id
       AND activity_participants.user_id = md5('user-2')::uuid
       AND activity_participants.status IN ('requested', 'approved')
      ORDER BY activities.starts_at DESC, activities.id DESC LIMIT 25
      """,
      "messages" => """
      SELECT id FROM messages
      WHERE assignment_id = md5('assignment-1')::uuid
      ORDER BY inserted_at DESC, id DESC LIMIT 51
      """,
      "activity_messages" => """
      SELECT id FROM activity_messages
      WHERE activity_id = md5('activity-1')::uuid
      ORDER BY inserted_at DESC, id DESC LIMIT 51
      """,
      "visible_reviews" => """
      SELECT id FROM reviews
      WHERE reviewee_id = md5('user-2')::uuid AND revealed_at IS NOT NULL
      ORDER BY inserted_at DESC, id DESC LIMIT 25
      """,
      "reports_open" => """
      SELECT id FROM reports
      WHERE status = 'open'
      ORDER BY inserted_at DESC, id DESC LIMIT 25
      """,
      "reports_all" => """
      SELECT id FROM reports
      ORDER BY inserted_at DESC, id DESC LIMIT 25
      """,
      "blocks" => """
      SELECT id FROM blocks
      WHERE blocker_id = md5('user-2')::uuid
      ORDER BY inserted_at DESC, id DESC LIMIT 25
      """,
      "abuse_signals" => """
      SELECT id FROM abuse_signals
      WHERE status = 'open'
      ORDER BY inserted_at DESC, id DESC LIMIT 25
      """,
      "proposals_open" => """
      SELECT id FROM category_proposals
      WHERE status = 'open'
      ORDER BY inserted_at DESC, id DESC LIMIT 25
      """,
      "proposals_all" => """
      SELECT id FROM category_proposals
      ORDER BY inserted_at DESC, id DESC LIMIT 25
      """,
      "users" => """
      SELECT id FROM users
      ORDER BY inserted_at DESC, id DESC LIMIT 25
      """,
      "reputation_helper" => """
      SELECT
        count(help_assignments.id),
        count(DISTINCT help_requests.requester_id),
        count(help_assignments.id) FILTER (
          WHERE help_assignments.handover_verified_at IS NOT NULL
        ),
        count(help_assignments.id) FILTER (
          WHERE help_assignments.proximity_observed_at IS NOT NULL
            AND help_assignments.helper_movement_observed_at IS NOT NULL
        )
      FROM help_assignments
      INNER JOIN help_requests
        ON help_requests.id = help_assignments.request_id
      WHERE help_assignments.status = 'completed'
        AND help_assignments.helper_id = md5('user-2')::uuid
      """,
      "leaderboard_aggregate" => """
      SELECT
        help_assignments.helper_id,
        count(help_assignments.id),
        count(DISTINCT help_requests.requester_id),
        count(DISTINCT help_requests.requester_id) FILTER (
          WHERE help_assignments.proximity_observed_at IS NOT NULL
            AND help_assignments.helper_movement_observed_at IS NOT NULL
        ),
        count(DISTINCT help_requests.requester_id) FILTER (
          WHERE help_assignments.handover_verified_at IS NOT NULL
        )
      FROM help_assignments
      INNER JOIN help_requests
        ON help_requests.id = help_assignments.request_id
      WHERE help_assignments.status = 'completed'
      GROUP BY help_assignments.helper_id
      """
    }
  end

  defp explain!(query) do
    result =
      Repo.query!("EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) #{query}", [], timeout: :infinity)

    case result.rows do
      [[[plan | _]]] when is_map(plan) -> plan
      [[plan]] when is_map(plan) -> plan
      rows -> Mix.raise("unexpected EXPLAIN JSON result: #{inspect(rows, limit: 2)}")
    end
  end

  defp summarize_plan(plan) do
    nodes = flatten_nodes(Map.fetch!(plan, "Plan"))

    %{
      planning_time_ms: plan["Planning Time"],
      execution_time_ms: plan["Execution Time"],
      node_types: nodes |> Enum.map(& &1["Node Type"]) |> Enum.reject(&is_nil/1),
      index_names: nodes |> Enum.map(& &1["Index Name"]) |> Enum.reject(&is_nil/1) |> Enum.uniq(),
      shared_hit_blocks: plan["Plan"]["Shared Hit Blocks"] || 0,
      shared_read_blocks: plan["Plan"]["Shared Read Blocks"] || 0,
      returned_rows: plan["Plan"]["Actual Rows"]
    }
  end

  defp flatten_nodes(node) do
    [node | Enum.flat_map(Map.get(node, "Plans", []), &flatten_nodes/1)]
  end

  defp expected_indexes do
    %{
      "help_discovery" => "help_requests_discovery_cursor_index",
      "help_requester" => "help_requests_requester_cursor_index",
      "activity_discovery" => "activities_discovery_cursor_index",
      "activity_participant" => "activity_participants_user_id_status_index",
      "messages" => "messages_assignment_cursor_index",
      "activity_messages" => "activity_messages_activity_cursor_index",
      "visible_reviews" => "reviews_visible_cursor_index",
      "reports_open" => "reports_cursor_index",
      "reports_all" => "reports_cursor_index",
      "blocks" => "blocks_blocker_cursor_index",
      "abuse_signals" => "abuse_signals_status_cursor_index",
      "proposals_open" => "category_proposals_cursor_index",
      "proposals_all" => "category_proposals_cursor_index",
      "users" => "users_moderation_cursor_index"
    }
  end

  defp assert_expected_indexes_exist! do
    existing =
      Repo.query!("SELECT indexname FROM pg_indexes WHERE schemaname = current_schema()").rows
      |> List.flatten()
      |> MapSet.new()

    missing =
      expected_indexes()
      |> Map.values()
      |> Enum.uniq()
      |> Enum.reject(&MapSet.member?(existing, &1))

    if missing != [], do: Mix.raise("missing cursor indexes: #{Enum.join(missing, ", ")}")
  end

  defp assert_expected_indexes_used!(plans) do
    failures =
      Enum.reject(expected_indexes(), fn {name, index} ->
        index in get_in(plans, [name, :index_names])
      end)

    if failures != [] do
      details =
        Enum.map_join(failures, ", ", fn {name, index} ->
          "#{name} expected #{index}, observed #{inspect(get_in(plans, [name, :index_names]))}"
        end)

      Mix.raise("cursor plan assertions failed: #{details}")
    end
  end

  defp assert_keyset_pages! do
    {first_ids, cursor} = fetch_help_page(nil)
    {second_ids, _cursor} = fetch_help_page(cursor)

    if length(first_ids) != 25 or length(second_ids) != 25 do
      Mix.raise("keyset page assertion expected two full 25-row pages")
    end

    if MapSet.disjoint?(MapSet.new(first_ids), MapSet.new(second_ids)) == false do
      Mix.raise("keyset page assertion found duplicate rows across pages")
    end

    expected =
      Repo.query!("""
      SELECT id FROM help_requests
      WHERE requester_id = md5('user-1')::uuid
      ORDER BY inserted_at DESC, id DESC LIMIT 50
      """).rows
      |> List.flatten()

    if first_ids ++ second_ids != expected do
      Mix.raise("keyset page assertion found a gap or order mismatch")
    end
  end

  defp fetch_help_page(nil) do
    result =
      Repo.query!("""
      SELECT id, inserted_at FROM help_requests
      WHERE requester_id = md5('user-1')::uuid
      ORDER BY inserted_at DESC, id DESC LIMIT 25
      """)

    page_result(result)
  end

  defp fetch_help_page({inserted_at, id}) do
    result =
      Repo.query!(
        """
        SELECT id, inserted_at FROM help_requests
        WHERE requester_id = md5('user-1')::uuid
          AND (inserted_at, id) < ($1, $2)
        ORDER BY inserted_at DESC, id DESC LIMIT 25
        """,
        [inserted_at, id]
      )

    page_result(result)
  end

  defp page_result(result) do
    ids = Enum.map(result.rows, &hd/1)
    [id, inserted_at] = List.last(result.rows)
    {ids, {inserted_at, id}}
  end

  defp assert_seed_size!(rows) do
    actual = scalar!("SELECT count(*) FROM help_requests")
    if actual != rows, do: Mix.raise("expected #{rows} seeded help requests, found #{actual}")
  end

  defp table_counts do
    ~w(
      users help_requests help_assignments messages reviews reports blocks
      abuse_signals category_proposals activities activity_participants activity_messages
    )
    |> Map.new(fn table -> {table, scalar!("SELECT count(*) FROM #{table}")} end)
  end

  defp index_sizes do
    Repo.query!("""
    SELECT indexrelname, pg_relation_size(indexrelid)
    FROM pg_stat_user_indexes
    WHERE indexrelname LIKE '%cursor_index'
    ORDER BY indexrelname
    """).rows
    |> Map.new(fn [name, bytes] -> {name, bytes} end)
  end

  defp write_comparison!(output, after_summary) do
    before_path = Path.join(output, "before-summary.json")

    before =
      before_path
      |> File.read!()
      |> Jason.decode!()

    comparisons =
      Map.new(after_summary.plans, fn {name, after_plan} ->
        before_plan = get_in(before, ["plans", name])

        {name,
         %{
           before_execution_time_ms: before_plan["execution_time_ms"],
           after_execution_time_ms: after_plan.execution_time_ms,
           observed_execution_ratio:
             ratio(before_plan["execution_time_ms"], after_plan.execution_time_ms),
           before_index_names: before_plan["index_names"],
           after_index_names: after_plan.index_names
         }}
      end)

    comparison = %{
      note:
        "Observed timings describe this isolated run and are not a portable minimum or latency gate.",
      configured_rows_per_large_table: after_summary.configured_rows_per_large_table,
      plans: comparisons
    }

    File.write!(
      Path.join(output, "comparison.json"),
      Jason.encode_to_iodata!(comparison, pretty: true)
    )

    markdown =
      [
        "# Isolated database scale measurement\n\n",
        "Rows configured per large table: ",
        Integer.to_string(after_summary.configured_rows_per_large_table),
        "\n\n",
        "Timings below are observations from this run, not a portable minimum or an SLO.\n\n",
        "| Query | Before, ms | After, ms | Before indexes | After indexes |\n",
        "| --- | ---: | ---: | --- | --- |\n",
        Enum.map(comparisons, fn {name, values} ->
          [
            "| ",
            name,
            " | ",
            format_ms(values.before_execution_time_ms),
            " | ",
            format_ms(values.after_execution_time_ms),
            " | ",
            Enum.join(values.before_index_names, ", "),
            " | ",
            Enum.join(values.after_index_names, ", "),
            " |\n"
          ]
        end)
      ]

    File.write!(Path.join(output, "README.md"), markdown)
  end

  defp ratio(before_ms, after_ms)
       when is_number(before_ms) and is_number(after_ms) and after_ms > 0,
       do: Float.round(before_ms / after_ms, 3)

  defp ratio(_before_ms, _after_ms), do: nil

  defp format_ms(value) when is_float(value), do: :erlang.float_to_binary(value, decimals: 3)
  defp format_ms(value), do: to_string(value)

  defp scalar!(sql) do
    case Repo.query!(sql).rows do
      [[value]] -> value
      rows -> Mix.raise("expected one scalar value, got: #{inspect(rows)}")
    end
  end

  defp query!(sql), do: Repo.query!(sql, [], timeout: :infinity)
end