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

  alias WhoNeedHelp.Accounts.{Scope, User}
  alias WhoNeedHelp.{Activities, Catalog, Help, Repo}
  alias WhoNeedHelp.Help.DiscoveryViewport

  @shortdoc "Prepares, appends, or verifies a persistent local discovery-scale data set"
  @isolated_confirmation "persistent-local-scale"
  @additive_confirmation "persistent-dev-scale"
  @default_rows 1_000_000
  @synthetic_users 1_000
  @viewer_email "scale-viewer@example.invalid"
  @fixture_version 6
  @request_open_modulus 100
  @activity_open_modulus 50
  @places [
    {"Kyiv", 30.5234, 50.4501},
    {"Lviv", 24.0316, 49.8429},
    {"Warsaw", 21.0122, 52.2297},
    {"Krakow", 19.945, 50.0647},
    {"Berlin", 13.405, 52.52},
    {"Hamburg", 9.9937, 53.5511},
    {"Frankfurt", 8.6821, 50.1109},
    {"Munich", 11.582, 48.1351},
    {"Prague", 14.4378, 50.0755},
    {"Vienna", 16.3738, 48.2082},
    {"Budapest", 19.0402, 47.4979},
    {"Bucharest", 26.1025, 44.4268},
    {"Sofia", 23.3219, 42.6977},
    {"Athens", 23.7275, 37.9838},
    {"Rome", 12.4964, 41.9028},
    {"Milan", 9.19, 45.4642},
    {"Madrid", -3.7038, 40.4168},
    {"Barcelona", 2.1734, 41.3851},
    {"Paris", 2.3522, 48.8566},
    {"Lyon", 4.8357, 45.764},
    {"London", -0.1276, 51.5072},
    {"Manchester", -2.2426, 53.4808},
    {"Dublin", -6.2603, 53.3498},
    {"Amsterdam", 4.9041, 52.3676},
    {"Brussels", 4.3517, 50.8503},
    {"Copenhagen", 12.5683, 55.6761},
    {"Stockholm", 18.0686, 59.3293},
    {"Oslo", 10.7522, 59.9139},
    {"Helsinki", 24.9384, 60.1699},
    {"Tallinn", 24.7536, 59.437},
    {"Vilnius", 25.2797, 54.6872},
    {"Riga", 24.1052, 56.9496},
    {"Istanbul", 28.9784, 41.0082},
    {"New York", -74.006, 40.7128},
    {"Boston", -71.0589, 42.3601},
    {"Washington", -77.0369, 38.9072},
    {"Chicago", -87.6298, 41.8781},
    {"Toronto", -79.3832, 43.6532},
    {"Montreal", -73.5673, 45.5017},
    {"Vancouver", -123.1207, 49.2827},
    {"Seattle", -122.3321, 47.6062},
    {"San Francisco", -122.4194, 37.7749},
    {"Los Angeles", -118.2437, 34.0522},
    {"Denver", -104.9903, 39.7392},
    {"Dallas", -96.797, 32.7767},
    {"Mexico City", -99.1332, 19.4326},
    {"Bogota", -74.0721, 4.711},
    {"Lima", -77.0428, -12.0464},
    {"Santiago", -70.6693, -33.4489},
    {"Buenos Aires", -58.3816, -34.6037},
    {"Sao Paulo", -46.6333, -23.5505},
    {"Rio de Janeiro", -43.1729, -22.9068},
    {"Montevideo", -56.1645, -34.9011},
    {"Cape Town", 18.4241, -33.9249},
    {"Johannesburg", 28.0473, -26.2041},
    {"Nairobi", 36.8219, -1.2921},
    {"Lagos", 3.3792, 6.5244},
    {"Accra", -0.187, 5.6037},
    {"Cairo", 31.2357, 30.0444},
    {"Casablanca", -7.5898, 33.5731},
    {"Delhi", 77.1025, 28.7041},
    {"Mumbai", 72.8777, 19.076},
    {"Bengaluru", 77.5946, 12.9716},
    {"Bangkok", 100.5018, 13.7563},
    {"Singapore", 103.8198, 1.3521},
    {"Jakarta", 106.8456, -6.2088},
    {"Manila", 120.9842, 14.5995},
    {"Hong Kong", 114.1694, 22.3193},
    {"Seoul", 126.978, 37.5665},
    {"Tokyo", 139.6917, 35.6895},
    {"Osaka", 135.5023, 34.6937},
    {"Taipei", 121.5654, 25.033},
    {"Sydney", 151.2093, -33.8688},
    {"Melbourne", 144.9631, -37.8136},
    {"Brisbane", 153.0251, -27.4698},
    {"Auckland", 174.7633, -36.8485},
    {"Tel Aviv", 34.7818, 32.0853},
    {"Dubai", 55.2708, 25.2048},
    {"Riyadh", 46.6753, 24.7136},
    {"Tbilisi", 44.793, 41.7151},
    {"Yerevan", 44.5152, 40.1872}
  ]
  @place_count length(@places)

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

    {options, rest, invalid} =
      OptionParser.parse(arguments,
        strict: [requests: :integer, activities: :integer, samples: :integer, output: :string]
      )

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

    action =
      case rest do
        [value] when value in ["prepare", "append", "verify"] -> value
        _other -> Mix.raise(usage())
      end

    request_count = positive_count!(options, :requests)
    activity_count = positive_count!(options, :activities)
    samples = positive_count!(options, :samples, 3)
    output = Keyword.get(options, :output) || Mix.raise("--output PATH is required")
    context = verified_context!(output, action)

    case action do
      "prepare" -> prepare!(context, request_count, activity_count)
      "append" -> append!(context, request_count, activity_count)
      "verify" -> verify!(context, request_count, activity_count, samples)
    end
  end

  defp usage do
    "usage: mix wnh.scale_fixtures prepare|append|verify " <>
      "[--requests COUNT] [--activities COUNT] [--samples COUNT] --output /output/FILE.json"
  end

  defp positive_count!(options, key, default \\ @default_rows) do
    value = Keyword.get(options, key, default)
    if is_integer(value) and value > 0, do: value, else: Mix.raise("--#{key} must be positive")
  end

  defp verified_context!(output, action) do
    expected_database = required_env!("WNH_SCALE_EXPECTED_DATABASE")
    output = Path.expand(output)

    expected_confirmation =
      if action == "append", do: @additive_confirmation, else: @isolated_confirmation

    unless System.get_env("WNH_SCALE_FIXTURE_CONFIRM") == expected_confirmation do
      Mix.raise("WNH_SCALE_FIXTURE_CONFIRM must equal #{expected_confirmation}")
    end

    unless String.starts_with?(output, "/output/") do
      Mix.raise("--output must resolve below /output")
    end

    %Postgrex.Result{rows: [[actual_database]]} =
      Repo.query!("SELECT current_database()", [], log: false)

    unless actual_database == expected_database do
      Mix.raise(
        "refusing scale mutation: expected database #{inspect(expected_database)}, " <>
          "observed #{inspect(actual_database)}"
      )
    end

    %{database: actual_database, output: output}
  end

  defp append!(context, request_count, activity_count) do
    before_totals = discovery_counts()
    before_synthetic = synthetic_discovery_counts()

    if before_synthetic.help_requests > request_count or
         before_synthetic.activities > activity_count do
      Mix.raise(
        "the development database already contains more synthetic scale rows than requested: " <>
          "#{before_synthetic.help_requests} requests and #{before_synthetic.activities} activities"
      )
    end

    categories = seed_categories!()
    ensure_viewer!()

    {:ok, profile} =
      Repo.transaction(
        fn ->
          seed_users!()
          seed_requests!(request_count, categories.help)
          seed_activities!(activity_count, categories.activity)
          assert_synthetic_counts!(request_count, activity_count)
          assert_fixture_profile!(request_count, activity_count)
        end,
        timeout: :infinity
      )

    analyze_scale_tables!()

    write_summary!(context, "appended", request_count, activity_count, %{
      expected_kind: "synthetic",
      totals_before: before_totals,
      synthetic_before: before_synthetic,
      synthetic_after: synthetic_discovery_counts(),
      fixture_profile: profile
    })

    totals = discovery_counts()

    Mix.shell().info(
      "persistent synthetic scale data is ready alongside existing data in #{context.database}: " <>
        "#{totals.help_requests} total requests and #{totals.activities} total activities"
    )
  end

  defp prepare!(context, request_count, activity_count) do
    counts = discovery_counts()

    cond do
      counts.help_requests == request_count and counts.activities == activity_count ->
        ensure_viewer!()

        profile =
          if current_fixture_version?() do
            assert_fixture_profile!(request_count, activity_count)
          else
            categories = seed_categories!()

            {:ok, profile} =
              Repo.transaction(
                fn ->
                  seed_users!()
                  seed_requests!(request_count, categories.help)
                  seed_activities!(activity_count, categories.activity)
                  assert_counts!(request_count, activity_count)
                  assert_fixture_profile!(request_count, activity_count)
                end,
                timeout: :infinity
              )

            analyze_scale_tables!()
            profile
          end

        write_summary!(context, "prepared", request_count, activity_count, %{
          fixture_profile: profile
        })

      counts.help_requests != 0 or counts.activities != 0 ->
        Mix.raise(
          "the isolated scale database already contains #{counts.help_requests} help requests " <>
            "and #{counts.activities} activities; refusing to mix data sets"
        )

      true ->
        categories = seed_categories!()
        ensure_viewer!()

        {:ok, profile} =
          Repo.transaction(
            fn ->
              seed_users!()
              seed_requests!(request_count, categories.help)
              seed_activities!(activity_count, categories.activity)
              assert_counts!(request_count, activity_count)
              assert_fixture_profile!(request_count, activity_count)
            end,
            timeout: :infinity
          )

        analyze_scale_tables!()

        write_summary!(context, "prepared", request_count, activity_count, %{
          fixture_profile: profile
        })
    end

    Mix.shell().info(
      "persistent scale data is ready in #{context.database}: " <>
        "#{request_count} requests and #{activity_count} activities"
    )
  end

  defp verify!(context, request_count, activity_count, samples) do
    assert_counts!(request_count, activity_count)
    assert_fixture_profile!(request_count, activity_count)
    viewer = Repo.get_by!(User, email: @viewer_email)
    scope = Scope.for_user(viewer)

    viewports = %{
      kyiv: viewport!(30.30, 50.25, 30.80, 50.65, 10),
      europe: viewport!(-12.0, 34.0, 35.0, 61.0, 4),
      world: viewport!(-179.99, -80.0, 179.99, 80.0, 1)
    }

    measurements =
      Map.new(viewports, fn {name, viewport} ->
        request_page =
          timed_samples(
            fn -> Help.paginate_open_requests(scope, %{}, viewport: viewport) end,
            samples
          )

        request_map =
          timed_samples(fn -> Help.map_discovery_items(scope, %{}, viewport) end, samples)

        activity_page =
          timed_samples(
            fn -> Activities.paginate_open_activities(scope, %{}, viewport: viewport) end,
            samples
          )

        activity_map =
          timed_samples(fn -> Activities.map_discovery_items(scope, %{}, viewport) end, samples)

        {name,
         %{
           request_page: page_measurement(request_page),
           request_map: map_measurement(request_map),
           activity_page: page_measurement(activity_page),
           activity_map: map_measurement(activity_map)
         }}
      end)

    write_summary!(context, "verified", request_count, activity_count, %{
      fixture_profile: fixture_profile(),
      measurement_samples: samples,
      viewport_measurements: measurements
    })

    Mix.shell().info("scale verification written to #{context.output}")
  end

  defp seed_categories! do
    Catalog.seed_defaults()

    categories = Catalog.list_all_categories()

    %{
      help: category_ids!(categories, :help),
      activity: category_ids!(categories, :activity)
    }
  end

  defp category_ids!(categories, mode) do
    leaves =
      Enum.filter(categories, fn category ->
        category.mode == mode and not is_nil(category.parent_id)
      end)

    ids =
      case leaves do
        [] -> categories |> Enum.filter(&(&1.mode == mode)) |> Enum.map(& &1.id)
        values -> Enum.map(values, & &1.id)
      end

    if ids == [], do: Mix.raise("no #{mode} categories were seeded"), else: ids
  end

  defp ensure_viewer! do
    password = required_env!("SCALE_VIEWER_PASSWORD")

    unless byte_size(password) in 12..72 do
      Mix.raise("SCALE_VIEWER_PASSWORD must contain between 12 and 72 bytes")
    end

    case Repo.get_by(User, email: @viewer_email) do
      nil ->
        now = DateTime.utc_now(:second)

        %User{
          email: @viewer_email,
          display_name: "Scale Viewer",
          hashed_password: Bcrypt.hash_pwd_salt(password),
          confirmed_at: now,
          accepted_terms_at: now,
          locale: "en"
        }
        |> Repo.insert!()

      %User{} ->
        :ok
    end
  end

  defp seed_users! do
    Repo.query!(
      """
      INSERT INTO users (
        id, email, display_name, confirmed_at, accepted_terms_at, locale,
        inserted_at, updated_at
      )
      SELECT
        md5('wnh-scale-user-' || value)::uuid,
        'scale-user-' || value || '@example.invalid',
        'Scale user ' || value,
        date_trunc('second', now()),
        date_trunc('second', now()),
        CASE value % 3 WHEN 0 THEN 'uk' WHEN 1 THEN 'en' ELSE 'ru' END,
        date_trunc('second', now()) - value * interval '1 second',
        date_trunc('second', now()) - value * interval '1 second'
      FROM generate_series(1, $1) AS value
      ON CONFLICT DO NOTHING
      """,
      [@synthetic_users],
      timeout: :infinity,
      log: false
    )
  end

  defp seed_requests!(rows, category_ids) do
    {place_names, place_longitudes, place_latitudes} = place_columns()

    Repo.query!(
      """
      INSERT INTO help_requests (
        id, title, description, pickup_instructions, structured_data,
        location_label, location, status, urgency, location_visibility,
        location_radius_meters, expires_at, requester_id, category_id,
        inserted_at, updated_at
      )
      SELECT
        md5('wnh-scale-request-v1-' || value)::uuid,
        'Scale help request ' || value,
        'Synthetic scale request used to test geographic discovery, privacy, and UI.',
        'Synthetic data only; no real-world action is requested.',
        jsonb_build_object(
          'scale_fixture', true,
          'scale_fixture_version', $7::integer
        ),
        place.name || ' synthetic area ' ||
          ((variation.category_key + variation.time_key) % 100 + 1),
        CASE WHEN variation.visibility_key % 20 = 0 THEN NULL
             ELSE ST_SetSRID(
               ST_MakePoint(coordinates.longitude, coordinates.latitude),
               4326
             )
        END,
        CASE WHEN variation.status_key % #{@request_open_modulus} = 0 THEN 'open'
             WHEN variation.status_key % #{@request_open_modulus} < 86 THEN 'completed'
             ELSE 'cancelled' END,
        CASE (variation.status_key / 10) % 3
          WHEN 0 THEN 'now'
          WHEN 1 THEN 'today'
          ELSE 'scheduled'
        END,
        CASE WHEN variation.visibility_key % 20 = 0 THEN 'hidden'
             WHEN variation.visibility_key % 20 < 5 THEN 'exact_public'
             WHEN variation.visibility_key % 20 < 9 THEN 'exact_for_active_match'
             ELSE 'approximate_public' END,
        CASE WHEN variation.visibility_key % 20 < 5 THEN NULL
             WHEN (variation.visibility_key / 20) % 3 = 0 THEN 500
             WHEN (variation.visibility_key / 20) % 3 = 1 THEN 1000
             ELSE 2000 END,
        CASE WHEN variation.status_key % #{@request_open_modulus} = 0
          THEN date_trunc('second', now()) +
            ((variation.time_key % 14) + 1) * interval '1 day'
          ELSE date_trunc('second', now()) -
            ((variation.time_key % 365) + 1) * interval '1 day'
        END,
        md5(
          'wnh-scale-user-' ||
            (variation.owner_key % $2 + 1)
        )::uuid,
        (($3::text[])[
          (variation.category_key % cardinality($3::text[])) + 1
        ])::uuid,
        date_trunc('second', now()) - value * interval '1 second',
        date_trunc('second', now()) - value * interval '1 second'
      FROM generate_series(1, $1) AS value
      CROSS JOIN LATERAL (
        SELECT
          md5('wnh-scale-request-v5-' || value) AS fixture_hash,
          md5('wnh-scale-request-geography-v5-' || value) AS geography_hash
      ) AS fixture
      CROSS JOIN LATERAL (
        SELECT
          ('x' || substr(fixture.fixture_hash, 1, 8))::bit(32)::bigint
            AS status_key,
          ('x' || substr(fixture.fixture_hash, 9, 8))::bit(32)::bigint
            AS visibility_key,
          ('x' || substr(fixture.fixture_hash, 17, 8))::bit(32)::bigint
            AS time_key,
          ('x' || substr(fixture.fixture_hash, 25, 8))::bit(32)::bigint
            AS category_key,
          ('x' || substr(fixture.geography_hash, 1, 8))::bit(32)::bigint
            AS place_key,
          ('x' || substr(fixture.geography_hash, 9, 8))::bit(32)::bigint
            AS angle_key,
          ('x' || substr(fixture.geography_hash, 17, 8))::bit(32)::bigint
            AS distance_key,
          ('x' || substr(fixture.geography_hash, 25, 8))::bit(32)::bigint
            AS owner_key
      ) AS variation
      CROSS JOIN LATERAL (
        SELECT name, longitude, latitude
        FROM unnest(
          $4::text[],
          $5::double precision[],
          $6::double precision[]
        ) WITH ORDINALITY AS places(name, longitude, latitude, position)
        WHERE places.position = variation.place_key % cardinality($4::text[]) + 1
      ) AS place
      CROSS JOIN LATERAL (
        SELECT
          2 * pi() *
            ((variation.angle_key % 1000003)::double precision / 1000003.0) AS angle,
          sqrt(
            ((variation.distance_key / 20 % 1000003)::double precision + 1) /
              1000004.0
          ) *
          CASE WHEN variation.distance_key % 20 < 12 THEN 15.0
               WHEN variation.distance_key % 20 < 18 THEN 60.0
               ELSE 220.0
          END AS distance_km
      ) AS geography
      CROSS JOIN LATERAL (
        SELECT
          place.latitude +
            geography.distance_km / 111.32 * sin(geography.angle) AS raw_latitude,
          place.longitude +
            geography.distance_km /
              (111.32 * greatest(abs(cos(radians(place.latitude))), 0.2)) *
              cos(geography.angle) AS raw_longitude
      ) AS raw_coordinates
      CROSS JOIN LATERAL (
        SELECT
          greatest(-84.0, least(84.0, raw_coordinates.raw_latitude)) AS latitude,
          (raw_coordinates.raw_longitude + 180.0) -
            360.0 * floor((raw_coordinates.raw_longitude + 180.0) / 360.0) -
            180.0 AS longitude
      ) AS coordinates
      ON CONFLICT (id) DO UPDATE SET
        title = EXCLUDED.title,
        description = EXCLUDED.description,
        pickup_instructions = EXCLUDED.pickup_instructions,
        structured_data = EXCLUDED.structured_data,
        location_label = EXCLUDED.location_label,
        location = EXCLUDED.location,
        status = EXCLUDED.status,
        urgency = EXCLUDED.urgency,
        location_visibility = EXCLUDED.location_visibility,
        location_radius_meters = EXCLUDED.location_radius_meters,
        expires_at = EXCLUDED.expires_at,
        requester_id = EXCLUDED.requester_id,
        category_id = EXCLUDED.category_id,
        updated_at = EXCLUDED.updated_at
      WHERE help_requests.structured_data @> '{"scale_fixture":true}'::jsonb
        AND help_requests.structured_data ->> 'scale_fixture_version'
          IS DISTINCT FROM $7::text
      """,
      [
        rows,
        @synthetic_users,
        category_ids,
        place_names,
        place_longitudes,
        place_latitudes,
        @fixture_version
      ],
      timeout: :infinity,
      log: false
    )
  end

  defp seed_activities!(rows, category_ids) do
    {place_names, place_longitudes, place_latitudes} = place_columns()

    Repo.query!(
      """
      INSERT INTO activities (
        id, title, description, structured_data, location_label, location,
        location_visibility, status, starts_at, join_deadline, capacity,
        creator_id, category_id, inserted_at, updated_at
      )
      SELECT
        md5('wnh-scale-activity-v1-' || value)::uuid,
        'Scale community activity ' || value,
        'Synthetic scale activity used to test geographic discovery, privacy, and UI.',
        jsonb_build_object(
          'scale_fixture', true,
          'scale_fixture_version', $7::integer
        ),
        place.name || ' synthetic area ' ||
          ((variation.category_key + variation.time_key) % 100 + 1),
        ST_SetSRID(
          ST_MakePoint(coordinates.longitude, coordinates.latitude),
          4326
        ),
        CASE WHEN variation.visibility_key % 10 = 0
          THEN 'hidden'
          ELSE 'approximate_public'
        END,
        CASE WHEN variation.status_key % #{@activity_open_modulus} = 0 THEN 'open'
             WHEN variation.status_key % #{@activity_open_modulus} < 41 THEN 'completed'
             ELSE 'cancelled' END,
        CASE WHEN variation.status_key % #{@activity_open_modulus} = 0
          THEN date_trunc('second', now()) +
            ((variation.time_key % 60) + 2) * interval '1 day'
          ELSE date_trunc('second', now()) -
            ((variation.time_key % 365) + 1) * interval '1 day'
        END,
        CASE WHEN variation.status_key % #{@activity_open_modulus} = 0
          THEN date_trunc('second', now()) +
            ((variation.time_key % 60) + 1) * interval '1 day'
          ELSE date_trunc('second', now()) -
            ((variation.time_key % 365) + 2) * interval '1 day'
        END,
        2 + ((variation.status_key + variation.time_key) % 19),
        md5(
          'wnh-scale-user-' ||
            (variation.owner_key % $2 + 1)
        )::uuid,
        (($3::text[])[
          (variation.category_key % cardinality($3::text[])) + 1
        ])::uuid,
        date_trunc('second', now()) - value * interval '1 second',
        date_trunc('second', now()) - value * interval '1 second'
      FROM generate_series(1, $1) AS value
      CROSS JOIN LATERAL (
        SELECT
          md5('wnh-scale-activity-v5-' || value) AS fixture_hash,
          md5('wnh-scale-activity-geography-v5-' || value) AS geography_hash
      ) AS fixture
      CROSS JOIN LATERAL (
        SELECT
          ('x' || substr(fixture.fixture_hash, 1, 8))::bit(32)::bigint
            AS status_key,
          ('x' || substr(fixture.fixture_hash, 9, 8))::bit(32)::bigint
            AS visibility_key,
          ('x' || substr(fixture.fixture_hash, 17, 8))::bit(32)::bigint
            AS time_key,
          ('x' || substr(fixture.fixture_hash, 25, 8))::bit(32)::bigint
            AS category_key,
          ('x' || substr(fixture.geography_hash, 1, 8))::bit(32)::bigint
            AS place_key,
          ('x' || substr(fixture.geography_hash, 9, 8))::bit(32)::bigint
            AS angle_key,
          ('x' || substr(fixture.geography_hash, 17, 8))::bit(32)::bigint
            AS distance_key,
          ('x' || substr(fixture.geography_hash, 25, 8))::bit(32)::bigint
            AS owner_key
      ) AS variation
      CROSS JOIN LATERAL (
        SELECT name, longitude, latitude
        FROM unnest(
          $4::text[],
          $5::double precision[],
          $6::double precision[]
        ) WITH ORDINALITY AS places(name, longitude, latitude, position)
        WHERE places.position = variation.place_key % cardinality($4::text[]) + 1
      ) AS place
      CROSS JOIN LATERAL (
        SELECT
          2 * pi() *
            ((variation.angle_key % 1000003)::double precision / 1000003.0) AS angle,
          sqrt(
            ((variation.distance_key / 20 % 1000003)::double precision + 1) /
              1000004.0
          ) *
          CASE WHEN variation.distance_key % 20 < 12 THEN 15.0
               WHEN variation.distance_key % 20 < 18 THEN 60.0
               ELSE 220.0
          END AS distance_km
      ) AS geography
      CROSS JOIN LATERAL (
        SELECT
          place.latitude +
            geography.distance_km / 111.32 * sin(geography.angle) AS raw_latitude,
          place.longitude +
            geography.distance_km /
              (111.32 * greatest(abs(cos(radians(place.latitude))), 0.2)) *
              cos(geography.angle) AS raw_longitude
      ) AS raw_coordinates
      CROSS JOIN LATERAL (
        SELECT
          greatest(-84.0, least(84.0, raw_coordinates.raw_latitude)) AS latitude,
          (raw_coordinates.raw_longitude + 180.0) -
            360.0 * floor((raw_coordinates.raw_longitude + 180.0) / 360.0) -
            180.0 AS longitude
      ) AS coordinates
      ON CONFLICT (id) DO UPDATE SET
        title = EXCLUDED.title,
        description = EXCLUDED.description,
        structured_data = EXCLUDED.structured_data,
        location_label = EXCLUDED.location_label,
        location = EXCLUDED.location,
        location_visibility = EXCLUDED.location_visibility,
        status = EXCLUDED.status,
        starts_at = EXCLUDED.starts_at,
        join_deadline = EXCLUDED.join_deadline,
        capacity = EXCLUDED.capacity,
        creator_id = EXCLUDED.creator_id,
        category_id = EXCLUDED.category_id,
        updated_at = EXCLUDED.updated_at
      WHERE activities.structured_data @> '{"scale_fixture":true}'::jsonb
        AND activities.structured_data ->> 'scale_fixture_version'
          IS DISTINCT FROM $7::text
      """,
      [
        rows,
        @synthetic_users,
        category_ids,
        place_names,
        place_longitudes,
        place_latitudes,
        @fixture_version
      ],
      timeout: :infinity,
      log: false
    )
  end

  defp place_columns do
    {
      Enum.map(@places, &elem(&1, 0)),
      Enum.map(@places, &elem(&1, 1)),
      Enum.map(@places, &elem(&1, 2))
    }
  end

  defp viewport!(west, south, east, north, zoom) do
    {:ok, viewport} =
      DiscoveryViewport.cast(%{
        "west" => west,
        "south" => south,
        "east" => east,
        "north" => north,
        "zoom" => zoom,
        "width" => 1_280,
        "height" => 720
      })

    viewport
  end

  defp timed_samples(callback, samples) do
    _warm_result = callback.()

    measurements =
      Enum.map(1..samples, fn _sample ->
        {microseconds, result} = :timer.tc(callback)
        %{elapsed_ms: Float.round(microseconds / 1_000, 3), result: result}
      end)

    elapsed = measurements |> Enum.map(& &1.elapsed_ms) |> Enum.sort()

    %{
      elapsed_ms: median(elapsed),
      min_elapsed_ms: hd(elapsed),
      max_elapsed_ms: List.last(elapsed),
      samples_ms: elapsed,
      result: measurements |> List.last() |> Map.fetch!(:result)
    }
  end

  defp median(values) do
    middle = div(length(values), 2)

    if rem(length(values), 2) == 1 do
      Enum.at(values, middle)
    else
      Float.round((Enum.at(values, middle - 1) + Enum.at(values, middle)) / 2, 3)
    end
  end

  defp page_measurement(%{result: page} = measurement) do
    measurement
    |> timing_measurement()
    |> Map.merge(%{
      returned_rows: length(page.entries),
      has_next_page: not is_nil(page.next_cursor)
    })
  end

  defp map_measurement(%{result: items} = measurement) do
    measurement
    |> timing_measurement()
    |> Map.merge(%{
      returned_rows: length(items),
      represented_records: Enum.reduce(items, 0, &(Map.get(&1, :count, 1) + &2)),
      json_bytes: items |> Jason.encode!() |> byte_size()
    })
  end

  defp timing_measurement(measurement) do
    Map.take(measurement, [:elapsed_ms, :min_elapsed_ms, :max_elapsed_ms, :samples_ms])
  end

  defp assert_counts!(request_count, activity_count) do
    actual = discovery_counts()

    unless actual.help_requests == request_count and actual.activities == activity_count do
      Mix.raise(
        "expected #{request_count} requests and #{activity_count} activities, " <>
          "found #{actual.help_requests} and #{actual.activities}"
      )
    end
  end

  defp assert_synthetic_counts!(request_count, activity_count) do
    actual = synthetic_discovery_counts()

    unless actual.help_requests == request_count and actual.activities == activity_count do
      Mix.raise(
        "expected #{request_count} synthetic requests and #{activity_count} synthetic activities, " <>
          "found #{actual.help_requests} and #{actual.activities}"
      )
    end
  end

  defp analyze_scale_tables! do
    Repo.query!(
      "ANALYZE users, categories, help_requests, activities",
      [],
      timeout: :infinity,
      log: false
    )
  end

  defp current_fixture_version? do
    counts = synthetic_discovery_counts()

    counts.help_requests > 0 and counts.activities > 0 and
      counts.help_requests == fixture_version_count!("help_requests") and
      counts.activities == fixture_version_count!("activities")
  end

  defp fixture_version_count!("help_requests") do
    scalar!("""
    SELECT count(*)
    FROM help_requests
    WHERE structured_data @> '{"scale_fixture": true}'::jsonb
      AND structured_data ->> 'scale_fixture_version' = '#{@fixture_version}'
    """)
  end

  defp fixture_version_count!("activities") do
    scalar!("""
    SELECT count(*)
    FROM activities
    WHERE structured_data @> '{"scale_fixture": true}'::jsonb
      AND structured_data ->> 'scale_fixture_version' = '#{@fixture_version}'
    """)
  end

  defp assert_fixture_profile!(request_count, activity_count) do
    unless current_fixture_version?() do
      Mix.raise("synthetic discovery rows are not all fixture version #{@fixture_version}")
    end

    profile = fixture_profile()

    if request_count >= 10_000 and activity_count >= 10_000 do
      assert_distribution_keys!(
        "request statuses",
        profile.help_requests.statuses,
        ~w(cancelled completed open)
      )

      assert_distribution_keys!(
        "request visibility modes",
        profile.help_requests.visibility,
        ~w(approximate_public exact_for_active_match exact_public hidden)
      )

      assert_distribution_keys!(
        "request radii",
        profile.help_requests.radii_meters,
        ~w(500 1000 2000 none)
      )

      assert_distribution_keys!(
        "activity statuses",
        profile.activities.statuses,
        ~w(cancelled completed open)
      )

      assert_distribution_keys!(
        "activity visibility modes",
        profile.activities.visibility,
        ~w(approximate_public hidden)
      )

      assert_geographic_profile!("help requests", profile.help_requests, request_count)
      assert_geographic_profile!("activities", profile.activities, activity_count)

      assert_open_profile!(
        "help requests",
        profile.help_requests,
        request_count,
        @request_open_modulus
      )

      assert_open_profile!(
        "activities",
        profile.activities,
        activity_count,
        @activity_open_modulus
      )
    end

    if profile.help_requests.privacy_invariant_violations != 0 do
      Mix.raise(
        "synthetic help requests contain " <>
          "#{profile.help_requests.privacy_invariant_violations} privacy invariant violations"
      )
    end

    if profile.activities.privacy_invariant_violations != 0 do
      Mix.raise(
        "synthetic activities contain " <>
          "#{profile.activities.privacy_invariant_violations} privacy invariant violations"
      )
    end

    profile
  end

  defp assert_distribution_keys!(label, distribution, expected_keys) do
    actual_keys = distribution |> Map.keys() |> Enum.sort()
    expected_keys = Enum.sort(expected_keys)

    unless actual_keys == expected_keys do
      Mix.raise("#{label} must contain #{inspect(expected_keys)}, found #{inspect(actual_keys)}")
    end
  end

  defp assert_geographic_profile!(label, profile, expected_count) do
    unless profile.place_count == @place_count do
      Mix.raise(
        "#{label} must cover all #{@place_count} fixture places, " <>
          "found #{profile.place_count}"
      )
    end

    if profile.max_rows_in_one_place * 50 > expected_count do
      Mix.raise(
        "#{label} are too concentrated: one fixture place contains " <>
          "#{profile.max_rows_in_one_place} of #{expected_count} rows"
      )
    end

    if profile.located_rows > 0 and
         profile.distinct_location_points * 100 < profile.located_rows * 98 do
      Mix.raise(
        "#{label} contain too many duplicate coordinates: " <>
          "#{profile.distinct_location_points} distinct points for " <>
          "#{profile.located_rows} located rows"
      )
    end

    unless profile.distinct_public_marker_points == profile.located_rows do
      Mix.raise(
        "#{label} contain repeated public map coordinates: " <>
          "#{profile.distinct_public_marker_points} distinct public markers for " <>
          "#{profile.located_rows} located rows"
      )
    end
  end

  defp assert_open_profile!(label, profile, expected_count, open_modulus) do
    expected_open_rows = div(expected_count, open_modulus)
    observed_open_rows = Map.fetch!(profile.statuses, "open")
    tolerance = max(div(expected_open_rows, 5), 20)

    if abs(observed_open_rows - expected_open_rows) > tolerance do
      Mix.raise(
        "#{label} open-row profile is outside the synthetic contract: " <>
          "expected about #{expected_open_rows}, found #{observed_open_rows}"
      )
    end

    if expected_count >= 100_000 do
      max_public_open_rows = div(expected_count * 2, @place_count * open_modulus) + 1

      if profile.max_public_open_rows_in_one_place > max_public_open_rows do
        Mix.raise(
          "#{label} public open rows are too dense in one fixture place: " <>
            "#{profile.max_public_open_rows_in_one_place} exceeds " <>
            "#{max_public_open_rows}"
        )
      end
    end
  end

  defp fixture_profile do
    %{
      fixture_version: @fixture_version,
      fixture_places: @place_count,
      help_requests: %{
        statuses:
          distribution!("""
          SELECT status, count(*)
          FROM help_requests
          WHERE structured_data @> '{"scale_fixture": true}'::jsonb
          GROUP BY status
          ORDER BY status
          """),
        visibility:
          distribution!("""
          SELECT location_visibility, count(*)
          FROM help_requests
          WHERE structured_data @> '{"scale_fixture": true}'::jsonb
          GROUP BY location_visibility
          ORDER BY location_visibility
          """),
        radii_meters:
          distribution!("""
          SELECT coalesce(location_radius_meters::text, 'none'), count(*)
          FROM help_requests
          WHERE structured_data @> '{"scale_fixture": true}'::jsonb
          GROUP BY location_radius_meters
          ORDER BY location_radius_meters NULLS FIRST
          """),
        place_count: fixture_place_count!("help_requests"),
        max_rows_in_one_place: max_rows_in_one_place!("help_requests"),
        public_open_rows:
          scalar!("""
          SELECT count(*)
          FROM help_requests
          WHERE structured_data @> '{"scale_fixture": true}'::jsonb
            AND status = 'open'
            AND expires_at > now()
            AND location_visibility <> 'hidden'
          """),
        max_public_open_rows_in_one_place: max_public_open_rows_in_one_place!("help_requests"),
        located_rows:
          scalar!("""
          SELECT count(*)
          FROM help_requests
          WHERE structured_data @> '{"scale_fixture": true}'::jsonb
            AND location IS NOT NULL
          """),
        distinct_location_points:
          scalar!("""
          SELECT count(DISTINCT ST_AsBinary(location))
          FROM help_requests
          WHERE structured_data @> '{"scale_fixture": true}'::jsonb
            AND location IS NOT NULL
          """),
        distinct_public_marker_points:
          scalar!("""
          SELECT count(DISTINCT ST_AsBinary(location))
          FROM help_requests
          WHERE structured_data @> '{"scale_fixture": true}'::jsonb
            AND location IS NOT NULL
          """),
        privacy_invariant_violations:
          scalar!("""
          SELECT count(*)
          FROM help_requests
          WHERE structured_data @> '{"scale_fixture": true}'::jsonb
            AND (
              (location_visibility = 'hidden' AND
                (location IS NOT NULL OR location_radius_meters IS NOT NULL))
              OR
              (location_visibility = 'exact_public' AND
                (location IS NULL OR location_radius_meters IS NOT NULL))
              OR
              (location_visibility IN ('approximate_public', 'exact_for_active_match') AND
                (location IS NULL OR location_radius_meters NOT IN (500, 1000, 2000)))
            )
          """)
      },
      activities: %{
        statuses:
          distribution!("""
          SELECT status, count(*)
          FROM activities
          WHERE structured_data @> '{"scale_fixture": true}'::jsonb
          GROUP BY status
          ORDER BY status
          """),
        visibility:
          distribution!("""
          SELECT location_visibility, count(*)
          FROM activities
          WHERE structured_data @> '{"scale_fixture": true}'::jsonb
          GROUP BY location_visibility
          ORDER BY location_visibility
          """),
        place_count: fixture_place_count!("activities"),
        max_rows_in_one_place: max_rows_in_one_place!("activities"),
        public_open_rows:
          scalar!("""
          SELECT count(*)
          FROM activities
          WHERE structured_data @> '{"scale_fixture": true}'::jsonb
            AND status = 'open'
            AND starts_at > now()
            AND location_visibility = 'approximate_public'
          """),
        max_public_open_rows_in_one_place: max_public_open_rows_in_one_place!("activities"),
        located_rows:
          scalar!("""
          SELECT count(*)
          FROM activities
          WHERE structured_data @> '{"scale_fixture": true}'::jsonb
            AND location IS NOT NULL
          """),
        distinct_location_points:
          scalar!("""
          SELECT count(DISTINCT ST_AsBinary(location))
          FROM activities
          WHERE structured_data @> '{"scale_fixture": true}'::jsonb
            AND location IS NOT NULL
          """),
        distinct_public_marker_points:
          scalar!("""
          SELECT count(DISTINCT ST_AsBinary(location))
          FROM activities
          WHERE structured_data @> '{"scale_fixture": true}'::jsonb
            AND location IS NOT NULL
          """),
        privacy_invariant_violations:
          scalar!("""
          SELECT count(*)
          FROM activities
          WHERE structured_data @> '{"scale_fixture": true}'::jsonb
            AND (
              location IS NULL
              OR location_visibility NOT IN ('approximate_public', 'hidden')
            )
          """)
      }
    }
  end

  defp fixture_place_count!("help_requests") do
    scalar!("""
    SELECT count(DISTINCT regexp_replace(location_label, ' synthetic area [0-9]+$', ''))
    FROM help_requests
    WHERE structured_data @> '{"scale_fixture": true}'::jsonb
    """)
  end

  defp fixture_place_count!("activities") do
    scalar!("""
    SELECT count(DISTINCT regexp_replace(location_label, ' synthetic area [0-9]+$', ''))
    FROM activities
    WHERE structured_data @> '{"scale_fixture": true}'::jsonb
    """)
  end

  defp max_rows_in_one_place!("help_requests") do
    scalar!("""
    SELECT coalesce(max(place_rows), 0)
    FROM (
      SELECT count(*) AS place_rows
      FROM help_requests
      WHERE structured_data @> '{"scale_fixture": true}'::jsonb
      GROUP BY regexp_replace(location_label, ' synthetic area [0-9]+$', '')
    ) AS grouped_places
    """)
  end

  defp max_rows_in_one_place!("activities") do
    scalar!("""
    SELECT coalesce(max(place_rows), 0)
    FROM (
      SELECT count(*) AS place_rows
      FROM activities
      WHERE structured_data @> '{"scale_fixture": true}'::jsonb
      GROUP BY regexp_replace(location_label, ' synthetic area [0-9]+$', '')
    ) AS grouped_places
    """)
  end

  defp max_public_open_rows_in_one_place!("help_requests") do
    scalar!("""
    SELECT coalesce(max(place_rows), 0)
    FROM (
      SELECT count(*) AS place_rows
      FROM help_requests
      WHERE structured_data @> '{"scale_fixture": true}'::jsonb
        AND status = 'open'
        AND expires_at > now()
        AND location_visibility <> 'hidden'
      GROUP BY regexp_replace(location_label, ' synthetic area [0-9]+$', '')
    ) AS grouped_places
    """)
  end

  defp max_public_open_rows_in_one_place!("activities") do
    scalar!("""
    SELECT coalesce(max(place_rows), 0)
    FROM (
      SELECT count(*) AS place_rows
      FROM activities
      WHERE structured_data @> '{"scale_fixture": true}'::jsonb
        AND status = 'open'
        AND starts_at > now()
        AND location_visibility = 'approximate_public'
      GROUP BY regexp_replace(location_label, ' synthetic area [0-9]+$', '')
    ) AS grouped_places
    """)
  end

  defp distribution!(sql) do
    Repo.query!(sql, [], timeout: :infinity, log: false).rows
    |> Map.new(fn [value, count] -> {to_string(value), count} end)
  end

  defp discovery_counts do
    %{
      help_requests: scalar!("SELECT count(*) FROM help_requests"),
      activities: scalar!("SELECT count(*) FROM activities")
    }
  end

  defp synthetic_discovery_counts do
    %{
      help_requests:
        scalar!(
          "SELECT count(*) FROM help_requests " <>
            "WHERE structured_data @> '{\"scale_fixture\": true}'::jsonb"
        ),
      activities:
        scalar!(
          "SELECT count(*) FROM activities " <>
            "WHERE structured_data @> '{\"scale_fixture\": true}'::jsonb"
        )
    }
  end

  defp write_summary!(context, status, request_count, activity_count, extra) do
    File.mkdir_p!(Path.dirname(context.output))

    summary =
      Map.merge(
        %{
          schema_version: 1,
          status: status,
          database: context.database,
          expected: %{help_requests: request_count, activities: activity_count},
          counts: discovery_counts(),
          database_bytes: scalar!("SELECT pg_database_size(current_database())"),
          relation_bytes: relation_sizes(),
          captured_at: DateTime.utc_now() |> DateTime.truncate(:second) |> DateTime.to_iso8601()
        },
        extra
      )

    File.write!(context.output, Jason.encode_to_iodata!(summary, pretty: true))
  end

  defp relation_sizes do
    Repo.query!(
      """
      SELECT relation, pg_total_relation_size(relation::regclass)
      FROM unnest(ARRAY['users', 'categories', 'help_requests', 'activities']) AS relation
      ORDER BY relation
      """,
      [],
      log: false
    ).rows
    |> Map.new(fn [name, bytes] -> {name, bytes} end)
  end

  defp scalar!(sql) do
    Repo.query!(sql, [], timeout: :infinity, log: false).rows |> hd() |> hd()
  end

  defp required_env!(name) do
    case System.get_env(name) do
      value when is_binary(value) and value != "" -> value
      _missing -> Mix.raise("#{name} is required")
    end
  end
end