From a4414f09e49e2faebe55b9fa3c67d8bc3b62e5ef Mon Sep 17 00:00:00 2001 From: SimpleTest Date: Sun, 19 Jul 2026 05:12:07 +0300 Subject: [PATCH] feat: add measured cursor pagination --- README.md | 14 + docs/performance.md | 54 + docs/verification.md | 19 +- lib/mix/tasks/wnh.db_scale_benchmark.ex | 706 +++++++++++ lib/who_need_help/accounts.ex | 26 +- lib/who_need_help/activities.ex | 98 +- lib/who_need_help/catalog.ex | 38 +- lib/who_need_help/help.ex | 46 +- lib/who_need_help/messaging.ex | 37 +- lib/who_need_help/pagination.ex | 72 ++ lib/who_need_help/trust.ex | 402 +++++-- .../live/activity_live/index.ex | 64 +- .../live/activity_live/show.ex | 52 +- .../live/category_proposal_live.ex | 27 +- .../live/leaderboard_live.ex | 28 +- lib/who_need_help_web/live/moderation_live.ex | 100 +- lib/who_need_help_web/live/profile_live.ex | 59 +- .../live/request_live/index.ex | 60 +- .../live/request_live/show.ex | 58 +- priv/gettext/default.pot | 829 ++++++------- priv/gettext/en/LC_MESSAGES/default.po | 829 ++++++------- priv/gettext/ru/LC_MESSAGES/default.po | 1046 ++++++++--------- priv/gettext/uk/LC_MESSAGES/default.po | 1046 ++++++++--------- ...19013320_add_keyset_pagination_indexes.exs | 44 + scripts/db-scale-benchmark.sh | 81 ++ test/who_need_help/activities_test.exs | 69 +- test/who_need_help/mutual_aid_flow_test.exs | 83 ++ test/who_need_help/pagination_test.exs | 32 + test/who_need_help/trust_safety_test.exs | 37 +- 29 files changed, 3935 insertions(+), 2121 deletions(-) create mode 100644 lib/mix/tasks/wnh.db_scale_benchmark.ex create mode 100644 lib/who_need_help/pagination.ex create mode 100644 priv/repo/migrations/20260719013320_add_keyset_pagination_indexes.exs create mode 100755 scripts/db-scale-benchmark.sh create mode 100644 test/who_need_help/pagination_test.exs diff --git a/README.md b/README.md index 60ee780..b86cdeb 100644 --- a/README.md +++ b/README.md @@ -156,6 +156,20 @@ production release image. Its generated database credentials are random and exist only for that run. The exact database volume, networks, temporary source snapshot, and one-run images are removed automatically. +The cursor-pagination database benchmark also creates a one-run Compose +project, random database credentials, and a separate PostgreSQL volume: + +```bash +./scripts/db-scale-benchmark.sh +``` + +It seeds 50,000 rows in each large benchmark table by default, captures +PostgreSQL 18 JSON `EXPLAIN (ANALYZE, BUFFERS)` plans before and after the +cursor-index migration, verifies two consecutive keyset pages for gaps and +duplicates, and removes its database project and volume. `DB_SCALE_ROWS` +changes the sample size; it is an experiment input, not a resource minimum. +Ignored evidence is written below `output/db-scale/`. + The browser E2E command creates a uniquely named, isolated Compose project with its own PostGIS volume, Mailpit instance, Traefik proxy, two web replicas, and two worker replicas: diff --git a/docs/performance.md b/docs/performance.md index 78c285d..a29ecc5 100644 --- a/docs/performance.md +++ b/docs/performance.md @@ -119,3 +119,57 @@ Stop the isolated containers without deleting their database volume: The stop script reads the actual `LOAD_PROJECT`, refuses the staging project name, and does not delete the volume. Volume deletion is intentionally not automated. + +## Isolated database scale and cursor plans + +Run the database-only before/after measurement with: + +```sh +./scripts/db-scale-benchmark.sh +``` + +The script builds the test image and creates a uniquely named Compose project, +random one-run PostgreSQL credentials, and a separate PostGIS volume. It +migrates only through `20260719004249`, seeds the isolated baseline, records +machine-readable PostgreSQL +`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON)` plans, applies the generated +`20260719013320_add_keyset_pagination_indexes` migration, and records the same +queries again. It also verifies that two 25-row keyset pages exactly equal the +first 50 ordered rows without duplicates or gaps. The exact project, network, +volume, and image are removed by its trap. + +`DB_SCALE_ROWS` controls the configured sample size and defaults to 50,000. +That default is a reproducible experiment input, not a minimum database size, +capacity claim, latency target, or production traffic model. The structural +gate checks the canonical sample's chosen cursor indexes and query correctness; +it does not fail on an arbitrary millisecond threshold. + +Observed locally on 2026-07-19 with PostgreSQL 18.4, the canonical run created +50,000 rows in each of users, requests, assignments, messages, reviews, +reports, abuse signals, proposals, activities, and activity messages; it +created 50,500 activity participants and 49,998 block rows. The isolated +database occupied 231,225,023 bytes after the cursor migration. + +| Query | Baseline observed | Cursor migration observed | Selected cursor index | +| --- | ---: | ---: | --- | +| Urgent-help discovery | 8.437 ms | 0.149 ms | `help_requests_discovery_cursor_index` | +| Requester history | 11.287 ms | 0.038 ms | `help_requests_requester_cursor_index` | +| Activity discovery | 8.760 ms | 0.061 ms | `activities_discovery_cursor_index` | +| Visible reviews | 8.997 ms | 0.039 ms | `reviews_visible_cursor_index` | +| All reports | 9.293 ms | 0.026 ms | `reports_cursor_index` | +| All category proposals | 6.776 ms | 0.028 ms | `category_proposals_cursor_index` | +| Moderation users | 5.805 ms | 0.032 ms | `users_moderation_cursor_index` | +| Blocks | 5.388 ms | 0.044 ms | `blocks_blocker_cursor_index` | + +These are measurements of one warm local run and must not be interpreted as an +SLO or portable speedup. The measurement also exposed redundant candidate +indexes and an `OR`-based Activity membership query. The final migration keeps +the indexes PostgreSQL selected, uses partial discovery indexes for open, +non-hidden records, and the application joins the existing participant +membership invariant directly. Personal reputation and the leaderboard now +aggregate in PostgreSQL instead of loading all completed assignments into the +BEAM; the leaderboard itself uses composite keyset pagination. + +Ignored evidence for the recorded run: + +- `output/db-scale/20260719020251-1612535/` diff --git a/docs/verification.md b/docs/verification.md index d768da3..8a521c8 100644 --- a/docs/verification.md +++ b/docs/verification.md @@ -23,15 +23,16 @@ results from product limits and unknown production properties. ## Reproducible checks -- The isolated `./scripts/quality.sh` run completed on 2026-07-19 with 153 - tests and 0 failures after the full localization changes +- The isolated Phoenix suite completed on 2026-07-19 with 161 + tests and 0 failures after cursor pagination, database aggregation, and the + full localization changes on Elixir 1.20.2 and Erlang/OTP 29.0.3. - `mix compile --force --warnings-as-errors` and `mix format --check-formatted`: passed against the same final source. - `./scripts/quality.sh` passed ShellCheck 0.11.0, Hadolint 2.14.0 at warning threshold, actionlint 1.7.12, all four Compose renders, Helm lint, Trivy source/rendered-manifest scanning, xref, Credo high-priority checks, Sobelow - strict/private checks, Hex audit, 153 Phoenix tests, both npm audits, and the + strict/private checks, Hex audit, 161 Phoenix tests, both npm audits, and the production-image vulnerability scan. The rendered Helm manifest and Debian 13.6 release image each reported zero HIGH/CRITICAL findings under the configured gates. @@ -83,6 +84,14 @@ results from product limits and unknown production properties. select/boolean values use locale maps with an English fallback; the compatibility `description` column remains as a fallback for pre-existing categories. +- The isolated database-scale harness migrated an empty PostgreSQL 18.4 + database to the exact pre-index version, seeded the configured canonical + sample, captured JSON `EXPLAIN (ANALYZE, BUFFERS)` plans before and after the + generated cursor-index migration, and removed its project and volume. The + final 50,000-row-per-large-table run selected every asserted cursor index; + two consecutive request pages matched the first 50 ordered rows with no gap + or duplicate. Exact observations and their non-SLO limitations are recorded + in `docs/performance.md`. - Browser verification used headed Chrome. The authenticated matched-request page rendered its chat, MapLibre marker, and live-location controls with no console errors or warnings after the foreground-service rollout. @@ -351,8 +360,8 @@ regression was repeated against the final source on 2026-07-18: availability model. - Load-test representative data and traffic, then set measured pool, resource, autoscaling, and action-limit policies. -- Add pagination or bounded loading where real measurements show that request, - chat, moderation, or leaderboard result sets require it. +- Repeat representative authenticated write, chat, tracking, and reconnect + load scenarios before deriving production capacity or autoscaling policy. - Publish jurisdiction-specific emergency contacts, privacy, retention, prohibited-items, and voluntary-payment guidance after legal review. - Create and configure a GitHub OAuth App, then exercise the real external diff --git a/lib/mix/tasks/wnh.db_scale_benchmark.ex b/lib/mix/tasks/wnh.db_scale_benchmark.ex new file mode 100644 index 0000000..e8e5bc2 --- /dev/null +++ b/lib/mix/tasks/wnh.db_scale_benchmark.ex @@ -0,0 +1,706 @@ +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 diff --git a/lib/who_need_help/accounts.ex b/lib/who_need_help/accounts.ex index 39dbd15..2f331a8 100644 --- a/lib/who_need_help/accounts.ex +++ b/lib/who_need_help/accounts.ex @@ -4,6 +4,7 @@ defmodule WhoNeedHelp.Accounts do """ import Ecto.Query, warn: false + alias WhoNeedHelp.Pagination alias WhoNeedHelp.Repo alias WhoNeedHelp.Accounts.{Scope, SocialIdentity, User, UserToken, UserNotifier} @@ -102,12 +103,22 @@ defmodule WhoNeedHelp.Accounts do def admin_authorized?(_user), do: false def list_users_for_moderation(%Scope{user: user}) do + paginate_users_for_moderation(%Scope{user: user}).entries + end + + def paginate_users_for_moderation(%Scope{user: user}, options \\ []) do if moderator_authorized?(user) do + limit = Pagination.limit(options) + cursor = Pagination.cursor(options) + User - |> order_by([user], asc: user.moderation_status, desc: user.inserted_at) + |> before_moderation_user(cursor) + |> order_by([user], desc: user.inserted_at, desc: user.id) + |> limit(^(limit + 1)) |> Repo.all() + |> Pagination.page(limit, &{&1.inserted_at, &1.id}) else - [] + %Pagination.Page{} end end @@ -445,6 +456,17 @@ defmodule WhoNeedHelp.Accounts do ## Token helper + defp before_moderation_user(query, nil), do: query + + defp before_moderation_user(query, {inserted_at, id}) do + where( + query, + [user], + user.inserted_at < ^inserted_at or + (user.inserted_at == ^inserted_at and user.id < ^id) + ) + end + defp update_user_and_delete_all_tokens(changeset) do Repo.transact(fn -> with {:ok, user} <- Repo.update(changeset) do diff --git a/lib/who_need_help/activities.ex b/lib/who_need_help/activities.ex index ab2c021..0206785 100644 --- a/lib/who_need_help/activities.ex +++ b/lib/who_need_help/activities.ex @@ -9,6 +9,7 @@ defmodule WhoNeedHelp.Activities do alias WhoNeedHelp.Activities.{Activity, Message, Participant} alias WhoNeedHelp.Catalog alias WhoNeedHelp.Catalog.Category + alias WhoNeedHelp.Pagination alias WhoNeedHelp.Repo alias WhoNeedHelp.Trust alias WhoNeedHelp.Trust.Block @@ -28,7 +29,13 @@ defmodule WhoNeedHelp.Activities do end def list_open_activities(%Scope{user: user}, filters \\ %{}) do + paginate_open_activities(%Scope{user: user}, filters).entries + end + + def paginate_open_activities(%Scope{user: user}, filters \\ %{}, options \\ []) do now = DateTime.utc_now(:second) + limit = Pagination.limit(options) + cursor = Pagination.cursor(options) Activity |> where( @@ -49,28 +56,68 @@ defmodule WhoNeedHelp.Activities do ) ) |> maybe_filter_category(filters["category_id"] || filters[:category_id]) - |> order_by([activity], asc: activity.starts_at) + |> after_open_activity(cursor) + |> order_by([activity], asc: activity.starts_at, asc: activity.id) + |> limit(^(limit + 1)) |> preload(category: :parent, creator: :social_identities, participants: :user) |> Repo.all() |> Enum.map(&public_activity/1) + |> Pagination.page(limit, &{&1.starts_at, &1.id}) end def list_my_activities(%Scope{user: user}) do - participant_ids = - from participant in Participant, - where: - participant.user_id == ^user.id and - participant.status in [:requested, :approved], - select: participant.activity_id + paginate_my_activities(%Scope{user: user}).entries + end + + def paginate_my_activities(%Scope{user: user}, options \\ []) do + limit = Pagination.limit(options) + cursor = Pagination.cursor(options) Activity - |> where( - [activity], - activity.creator_id == ^user.id or activity.id in subquery(participant_ids) + |> join(:inner, [activity], participant in Participant, + on: + participant.activity_id == activity.id and participant.user_id == ^user.id and + participant.status in [:requested, :approved] ) - |> order_by([activity], desc: activity.starts_at) + |> before_my_activity(cursor) + |> order_by([activity], desc: activity.starts_at, desc: activity.id) + |> limit(^(limit + 1)) |> preload(category: :parent, participants: :user) |> Repo.all() + |> Pagination.page(limit, &{&1.starts_at, &1.id}) + end + + defp after_open_activity(query, nil), do: query + + defp after_open_activity(query, {starts_at, id}) do + where( + query, + [activity], + activity.starts_at > ^starts_at or + (activity.starts_at == ^starts_at and activity.id > ^id) + ) + end + + defp before_my_activity(query, nil), do: query + + defp before_my_activity(query, {starts_at, id}) do + where( + query, + [activity], + activity.starts_at < ^starts_at or + (activity.starts_at == ^starts_at and activity.id < ^id) + ) + end + + defp before_message(query, nil), do: query + + defp before_message(query, {inserted_at, id}) do + where( + query, + [message], + message.inserted_at < ^inserted_at or + (message.inserted_at == ^inserted_at and message.id < ^id) + ) end def get_activity(%Scope{user: user}, id) do @@ -103,6 +150,27 @@ defmodule WhoNeedHelp.Activities do end end + def paginate_messages(%Scope{user: user}, %Activity{} = activity, options \\ []) do + if activity.creator_id == user.id or approved_participant?(activity, user.id) do + limit = Pagination.limit(options, 50) + cursor = Pagination.cursor(options) + + page = + Message + |> where([message], message.activity_id == ^activity.id) + |> before_message(cursor) + |> order_by([message], desc: message.inserted_at, desc: message.id) + |> limit(^(limit + 1)) + |> preload(:sender) + |> Repo.all() + |> Pagination.page(limit, &{&1.inserted_at, &1.id}) + + %{page | entries: Enum.reverse(page.entries)} + else + %Pagination.Page{} + end + end + def change_activity(%Activity{} = activity, attrs \\ %{}) do activity |> Activity.create_changeset(attrs) @@ -491,13 +559,19 @@ defmodule WhoNeedHelp.Activities do end defp load_activity(id) do + recent_messages = + from message in Message, + order_by: [desc: message.inserted_at, desc: message.id], + limit: 50, + preload: :sender + Activity |> Repo.get!(id) |> Repo.preload( category: :parent, creator: :social_identities, participants: [user: :social_identities], - messages: :sender + messages: recent_messages ) end diff --git a/lib/who_need_help/catalog.ex b/lib/who_need_help/catalog.ex index d5912e0..efd4626 100644 --- a/lib/who_need_help/catalog.ex +++ b/lib/who_need_help/catalog.ex @@ -5,6 +5,7 @@ defmodule WhoNeedHelp.Catalog do alias WhoNeedHelp.Accounts alias WhoNeedHelp.Accounts.Scope alias WhoNeedHelp.Catalog.{Category, CategoryProposal, CategoryVote} + alias WhoNeedHelp.Pagination alias WhoNeedHelp.Repo alias WhoNeedHelp.Trust @@ -33,11 +34,21 @@ defmodule WhoNeedHelp.Catalog do def category_path(%Category{} = category, locale), do: Category.name(category, locale) def list_proposals do + paginate_proposals().entries + end + + def paginate_proposals(options \\ []) do + limit = Pagination.limit(options) + cursor = Pagination.cursor(options) + CategoryProposal |> where([p], p.status == :open) |> preload([:proposer, :parent, :votes]) - |> order_by([p], desc: p.inserted_at) + |> before_proposal(cursor) + |> order_by([proposal], desc: proposal.inserted_at, desc: proposal.id) + |> limit(^(limit + 1)) |> Repo.all() + |> Pagination.page(limit, &{&1.inserted_at, &1.id}) end def change_proposal(%CategoryProposal{} = proposal, attrs \\ %{}) do @@ -140,13 +151,23 @@ defmodule WhoNeedHelp.Catalog do def validate_structured_data(_category, _data), do: {:error, ["must be an object"]} def list_proposals_for_moderation(%Scope{user: user}) do + paginate_proposals_for_moderation(%Scope{user: user}).entries + end + + def paginate_proposals_for_moderation(%Scope{user: user}, options \\ []) do if Accounts.moderator_authorized?(user) do + limit = Pagination.limit(options) + cursor = Pagination.cursor(options) + CategoryProposal - |> order_by([proposal], asc: proposal.status, desc: proposal.inserted_at) + |> before_proposal(cursor) + |> order_by([proposal], desc: proposal.inserted_at, desc: proposal.id) + |> limit(^(limit + 1)) |> preload([:proposer, :parent, :merged_into, :reviewed_by, :votes]) |> Repo.all() + |> Pagination.page(limit, &{&1.inserted_at, &1.id}) else - [] + %Pagination.Page{} end end @@ -812,6 +833,17 @@ defmodule WhoNeedHelp.Catalog do ] end + defp before_proposal(query, nil), do: query + + defp before_proposal(query, {inserted_at, id}) do + where( + query, + [proposal], + proposal.inserted_at < ^inserted_at or + (proposal.inserted_at == ^inserted_at and proposal.id < ^id) + ) + end + defp names(en, uk, ru), do: %{"en" => en, "uk" => uk, "ru" => ru} defp option(value, label), do: %{"value" => value, "label" => label} diff --git a/lib/who_need_help/help.ex b/lib/who_need_help/help.ex index feb7e47..5cc61b3 100644 --- a/lib/who_need_help/help.ex +++ b/lib/who_need_help/help.ex @@ -8,6 +8,7 @@ defmodule WhoNeedHelp.Help do alias WhoNeedHelp.Catalog alias WhoNeedHelp.Catalog.Category alias WhoNeedHelp.Help.{Assignment, HelpRequest} + alias WhoNeedHelp.Pagination alias WhoNeedHelp.Repo alias WhoNeedHelp.Trust alias WhoNeedHelp.Trust.Block @@ -26,7 +27,13 @@ defmodule WhoNeedHelp.Help do end def list_open_requests(%Scope{user: user}, filters \\ %{}) do + paginate_open_requests(%Scope{user: user}, filters).entries + end + + def paginate_open_requests(%Scope{user: user}, filters \\ %{}, options \\ []) do now = DateTime.utc_now(:second) + limit = Pagination.limit(options) + cursor = Pagination.cursor(options) HelpRequest |> where( @@ -46,19 +53,54 @@ defmodule WhoNeedHelp.Help do ) ) |> filter_open_requests(filters) - |> order_by([r], asc: r.expires_at) + |> after_open_request(cursor) + |> order_by([request], asc: request.expires_at, asc: request.id) + |> limit(^(limit + 1)) |> preload([:category, :requester, assignment: :helper]) |> Repo.all() + |> Pagination.page(limit, &{&1.expires_at, &1.id}) end def list_open_requests, do: raise(ArgumentError, "an authenticated scope is required") def list_my_requests(%Scope{user: user}) do + paginate_my_requests(%Scope{user: user}).entries + end + + def paginate_my_requests(%Scope{user: user}, options \\ []) do + limit = Pagination.limit(options) + cursor = Pagination.cursor(options) + HelpRequest |> where([r], r.requester_id == ^user.id) - |> order_by([r], desc: r.inserted_at) + |> before_my_request(cursor) + |> order_by([request], desc: request.inserted_at, desc: request.id) + |> limit(^(limit + 1)) |> preload([:category, :requester, assignment: :helper]) |> Repo.all() + |> Pagination.page(limit, &{&1.inserted_at, &1.id}) + end + + defp after_open_request(query, nil), do: query + + defp after_open_request(query, {expires_at, id}) do + where( + query, + [request], + request.expires_at > ^expires_at or + (request.expires_at == ^expires_at and request.id > ^id) + ) + end + + defp before_my_request(query, nil), do: query + + defp before_my_request(query, {inserted_at, id}) do + where( + query, + [request], + request.inserted_at < ^inserted_at or + (request.inserted_at == ^inserted_at and request.id < ^id) + ) end def get_request!(id) do diff --git a/lib/who_need_help/messaging.ex b/lib/who_need_help/messaging.ex index eb4a0b1..fa1014c 100644 --- a/lib/who_need_help/messaging.ex +++ b/lib/who_need_help/messaging.ex @@ -6,6 +6,7 @@ defmodule WhoNeedHelp.Messaging do alias WhoNeedHelp.Help alias WhoNeedHelp.Help.Assignment alias WhoNeedHelp.Messaging.Message + alias WhoNeedHelp.Pagination alias WhoNeedHelp.Repo alias WhoNeedHelp.Trust @@ -14,15 +15,28 @@ defmodule WhoNeedHelp.Messaging do end def list_messages(%Scope{} = scope, %Assignment{} = assignment) do + paginate_messages(scope, assignment).entries + end + + def paginate_messages(%Scope{} = scope, %Assignment{} = assignment, options \\ []) do if Trust.eligible?(scope) and Help.participant?(scope, assignment) and not blocked_assignment?(scope, assignment) do - Message - |> where([m], m.assignment_id == ^assignment.id) - |> order_by([m], asc: m.inserted_at) - |> preload(:sender) - |> Repo.all() + limit = Pagination.limit(options, 50) + cursor = Pagination.cursor(options) + + page = + Message + |> where([message], message.assignment_id == ^assignment.id) + |> before_message(cursor) + |> order_by([message], desc: message.inserted_at, desc: message.id) + |> limit(^(limit + 1)) + |> preload(:sender) + |> Repo.all() + |> Pagination.page(limit, &{&1.inserted_at, &1.id}) + + %{page | entries: Enum.reverse(page.entries)} else - [] + %Pagination.Page{} end end @@ -61,4 +75,15 @@ defmodule WhoNeedHelp.Messaging do Trust.blocked_between?(user.id, counterpart_id) end + + defp before_message(query, nil), do: query + + defp before_message(query, {inserted_at, id}) do + where( + query, + [message], + message.inserted_at < ^inserted_at or + (message.inserted_at == ^inserted_at and message.id < ^id) + ) + end end diff --git a/lib/who_need_help/pagination.ex b/lib/who_need_help/pagination.ex new file mode 100644 index 0000000..d2cd741 --- /dev/null +++ b/lib/who_need_help/pagination.ex @@ -0,0 +1,72 @@ +defmodule WhoNeedHelp.Pagination do + @moduledoc """ + Small keyset-pagination primitives shared by database-backed lists. + + Cursors contain only the final row's UTC timestamp and UUID. They are + validated before being used as query parameters and carry no authorization + decision. + """ + + @default_limit 24 + @maximum_limit 100 + + defmodule Page do + @moduledoc false + defstruct entries: [], next_cursor: nil + end + + def limit(options, default \\ @default_limit) do + options + |> Keyword.get(:limit, default) + |> normalize_limit(default) + end + + def cursor(options) do + case decode(Keyword.get(options, :after)) do + {:ok, cursor} -> cursor + :error -> nil + end + end + + def page(rows, limit, cursor_fields) when is_function(cursor_fields, 1) do + entries = Enum.take(rows, limit) + + next_cursor = + if length(rows) > limit do + entries + |> List.last() + |> cursor_fields.() + |> then(fn {timestamp, id} -> encode(timestamp, id) end) + end + + %Page{entries: entries, next_cursor: next_cursor} + end + + def encode(%DateTime{} = timestamp, id) do + %{"at" => DateTime.to_iso8601(timestamp), "id" => to_string(id)} + |> Jason.encode!() + |> Base.url_encode64(padding: false) + end + + def decode(nil), do: {:ok, nil} + def decode(""), do: {:ok, nil} + + def decode(encoded) when is_binary(encoded) do + with {:ok, json} <- Base.url_decode64(encoded, padding: false), + {:ok, %{"at" => timestamp, "id" => id}} <- Jason.decode(json), + {:ok, timestamp, 0} <- DateTime.from_iso8601(timestamp), + {:ok, id} <- Ecto.UUID.cast(id) do + {:ok, {timestamp, id}} + else + _ -> :error + end + end + + def decode(_encoded), do: :error + + defp normalize_limit(limit, _default) + when is_integer(limit) and limit > 0 and limit <= @maximum_limit, + do: limit + + defp normalize_limit(_limit, default), do: min(default, @maximum_limit) +end diff --git a/lib/who_need_help/trust.ex b/lib/who_need_help/trust.ex index 777f4f5..2bed1e7 100644 --- a/lib/who_need_help/trust.ex +++ b/lib/who_need_help/trust.ex @@ -11,6 +11,7 @@ defmodule WhoNeedHelp.Trust do alias WhoNeedHelp.Help alias WhoNeedHelp.Help.{Assignment, HelpRequest} alias WhoNeedHelp.Messaging.Message + alias WhoNeedHelp.Pagination alias WhoNeedHelp.Repo alias WhoNeedHelp.Tracking.{Position, TrackingSession} @@ -100,80 +101,103 @@ defmodule WhoNeedHelp.Trust do end def visible_reviews(user_id) do + paginate_visible_reviews(user_id).entries + end + + def paginate_visible_reviews(user_id, options \\ []) do + limit = Pagination.limit(options) + cursor = Pagination.cursor(options) + Review |> where([review], review.reviewee_id == ^user_id and not is_nil(review.revealed_at)) - |> order_by([review], desc: review.inserted_at) + |> before_review(cursor) + |> order_by([review], desc: review.inserted_at, desc: review.id) + |> limit(^(limit + 1)) |> preload(:reviewer) |> Repo.all() + |> Pagination.page(limit, &{&1.inserted_at, &1.id}) end def reputation(user_id) do - rows = completed_rows() |> Enum.filter(&participant?(&1, user_id)) - counterparts = Enum.map(rows, &counterpart(&1, user_id)) - ratings = ratings(user_id) + helper_rows = + Assignment + |> join(:inner, [assignment], request in HelpRequest, + on: request.id == assignment.request_id + ) + |> where([assignment], assignment.status == :completed and assignment.helper_id == ^user_id) + |> select([assignment, request], %{ + id: assignment.id, + counterpart_id: request.requester_id, + verified: not is_nil(assignment.handover_verified_at), + location_supported: + not is_nil(assignment.proximity_observed_at) and + not is_nil(assignment.helper_movement_observed_at) + }) - %{ - completed: length(rows), - unique_people: counterparts |> Enum.uniq() |> length(), - verified_handovers: Enum.count(rows, &(not is_nil(&1.handover_verified_at))), - location_supported: - Enum.count( - rows, - &(not is_nil(&1.proximity_observed_at) and movement_for_user?(&1, user_id)) - ), - rating: average(ratings) - } + requester_rows = + Assignment + |> join(:inner, [assignment], request in HelpRequest, + on: request.id == assignment.request_id + ) + |> where( + [assignment, request], + assignment.status == :completed and request.requester_id == ^user_id and + assignment.helper_id != ^user_id + ) + |> select([assignment], %{ + id: assignment.id, + counterpart_id: assignment.helper_id, + verified: not is_nil(assignment.handover_verified_at), + location_supported: not is_nil(assignment.proximity_observed_at) + }) + + aggregate = + helper_rows + |> union_all(^requester_rows) + |> subquery() + |> select([row], %{ + completed: count(row.id), + unique_people: count(row.counterpart_id, :distinct), + verified_handovers: filter(count(row.id), row.verified), + location_supported: filter(count(row.id), row.location_supported) + }) + |> Repo.one!() + + Map.put(aggregate, :rating, average_rating(user_id)) end def leaderboard do - ratings_by_user = - Review - |> where([review], not is_nil(review.revealed_at)) - |> group_by([review], review.reviewee_id) - |> select([review], {review.reviewee_id, avg(review.rating)}) - |> Repo.all() - |> Map.new() + paginate_leaderboard().entries + end - completed_rows() - |> Enum.group_by(& &1.helper_id) - |> Enum.map(fn {helper_id, rows} -> - helper = rows |> hd() |> Map.fetch!(:helper) + def paginate_leaderboard(options \\ []) do + limit = Pagination.limit(options) + cursor = decode_leaderboard_cursor(Keyword.get(options, :after)) + rows = leaderboard_rows(cursor, limit) - supported_people = - rows - |> Enum.filter( - &(not is_nil(&1.proximity_observed_at) and - not is_nil(&1.helper_movement_observed_at)) - ) - |> Enum.map(& &1.requester_id) - |> Enum.uniq() - |> length() + page_rows = Enum.take(rows, limit) + users_by_id = users_by_id(Enum.map(page_rows, & &1.helper_id)) - verified_people = - rows - |> Enum.filter(&(not is_nil(&1.handover_verified_at))) - |> Enum.map(& &1.requester_id) - |> Enum.uniq() - |> length() + entries = + Enum.map(page_rows, fn row -> + %{ + user: Map.fetch!(users_by_id, row.helper_id), + completed: row.completed, + unique_people: row.unique_people, + location_supported_people: row.location_supported_people, + verified_people: row.verified_people, + rating: decimal_average(row.rating) + } + end) - %{ - user: helper, - completed: length(rows), - unique_people: rows |> Enum.map(& &1.requester_id) |> Enum.uniq() |> length(), - location_supported_people: supported_people, - verified_people: verified_people, - rating: ratings_by_user |> Map.get(helper_id) |> decimal_average() - } - end) - |> Enum.sort_by(fn entry -> - { - -entry.location_supported_people, - -entry.verified_people, - -entry.unique_people, - -entry.completed, - String.downcase(entry.user.display_name || "") - } - end) + next_cursor = + if length(rows) > limit do + page_rows + |> List.last() + |> encode_leaderboard_cursor() + end + + %Pagination.Page{entries: entries, next_cursor: next_cursor} end def report(%Scope{user: user} = scope, attrs) do @@ -197,10 +221,19 @@ defmodule WhoNeedHelp.Trust do end def list_reports(%Scope{user: user}, status \\ nil) do + paginate_reports(%Scope{user: user}, status).entries + end + + def paginate_reports(%Scope{user: user}, status \\ nil, options \\ []) do if Accounts.moderator_authorized?(user) do + limit = Pagination.limit(options) + cursor = Pagination.cursor(options) + Report |> maybe_status(status) - |> order_by([report], asc: report.status, desc: report.inserted_at) + |> before_report(cursor) + |> order_by([report], desc: report.inserted_at, desc: report.id) + |> limit(^(limit + 1)) |> preload([ :reporter, :reviewed_by, @@ -211,8 +244,9 @@ defmodule WhoNeedHelp.Trust do activity_message: :sender ]) |> Repo.all() + |> Pagination.page(limit, &{&1.inserted_at, &1.id}) else - [] + %Pagination.Page{} end end @@ -269,9 +303,11 @@ defmodule WhoNeedHelp.Trust do if assignment_id do Message |> where([message], message.assignment_id == ^assignment_id) - |> order_by([message], asc: message.inserted_at) + |> order_by([message], desc: message.inserted_at, desc: message.id) + |> limit(200) |> preload(:sender) |> Repo.all() + |> Enum.reverse() else [] end @@ -286,9 +322,11 @@ defmodule WhoNeedHelp.Trust do if activity_id do ActivityMessage |> where([message], message.activity_id == ^activity_id) - |> order_by([message], asc: message.inserted_at) + |> order_by([message], desc: message.inserted_at, desc: message.id) + |> limit(200) |> preload(:sender) |> Repo.all() + |> Enum.reverse() else [] end @@ -345,11 +383,21 @@ defmodule WhoNeedHelp.Trust do end def list_blocks(%Scope{user: user}) do + paginate_blocks(%Scope{user: user}).entries + end + + def paginate_blocks(%Scope{user: user}, options \\ []) do + limit = Pagination.limit(options) + cursor = Pagination.cursor(options) + Block |> where([block], block.blocker_id == ^user.id) - |> order_by([block], desc: block.inserted_at) + |> before_block(cursor) + |> order_by([block], desc: block.inserted_at, desc: block.id) + |> limit(^(limit + 1)) |> preload(:blocked) |> Repo.all() + |> Pagination.page(limit, &{&1.inserted_at, &1.id}) end def blocked_between?(first_user_id, second_user_id) do @@ -429,14 +477,24 @@ defmodule WhoNeedHelp.Trust do end def list_abuse_signals(%Scope{user: user}, status \\ :open) do + paginate_abuse_signals(%Scope{user: user}, status).entries + end + + def paginate_abuse_signals(%Scope{user: user}, status \\ :open, options \\ []) do if Accounts.moderator_authorized?(user) do + limit = Pagination.limit(options) + cursor = Pagination.cursor(options) + AbuseSignal |> where([signal], signal.status == ^status) - |> order_by([signal], desc: signal.inserted_at) + |> before_abuse_signal(cursor) + |> order_by([signal], desc: signal.inserted_at, desc: signal.id) + |> limit(^(limit + 1)) |> preload([:subject, :assignment, :reviewed_by]) |> Repo.all() + |> Pagination.page(limit, &{&1.inserted_at, &1.id}) else - [] + %Pagination.Page{} end end @@ -627,44 +685,204 @@ defmodule WhoNeedHelp.Trust do |> Repo.insert() end - defp completed_rows do - Assignment - |> join(:inner, [assignment], request in HelpRequest, on: request.id == assignment.request_id) - |> join(:inner, [assignment, _request], helper in User, on: helper.id == assignment.helper_id) - |> where([assignment], assignment.status == :completed) - |> select([assignment, request, helper], %{ - id: assignment.id, - helper_id: assignment.helper_id, - requester_id: request.requester_id, - handover_verified_at: assignment.handover_verified_at, - proximity_observed_at: assignment.proximity_observed_at, - helper_movement_observed_at: assignment.helper_movement_observed_at, - helper: helper + defp average_rating(user_id) do + Review + |> where([review], review.reviewee_id == ^user_id and not is_nil(review.revealed_at)) + |> select([review], avg(review.rating)) + |> Repo.one() + |> decimal_average() + end + + defp users_by_id([]), do: %{} + + defp users_by_id(ids) do + User + |> where([user], user.id in ^ids) + |> Repo.all() + |> Map.new(&{&1.id, &1}) + end + + defp leaderboard_rows(cursor, limit) do + stats = + Assignment + |> join(:inner, [assignment], request in HelpRequest, + on: request.id == assignment.request_id + ) + |> where([assignment], assignment.status == :completed) + |> group_by([assignment], assignment.helper_id) + |> select([assignment, request], %{ + helper_id: assignment.helper_id, + completed: count(assignment.id), + unique_people: count(request.requester_id, :distinct), + location_supported_people: + filter( + count(request.requester_id, :distinct), + not is_nil(assignment.proximity_observed_at) and + not is_nil(assignment.helper_movement_observed_at) + ), + verified_people: + filter( + count(request.requester_id, :distinct), + not is_nil(assignment.handover_verified_at) + ) + }) + + ratings = + Review + |> where([review], not is_nil(review.revealed_at)) + |> group_by([review], review.reviewee_id) + |> select([review], %{reviewee_id: review.reviewee_id, rating: avg(review.rating)}) + + stats + |> subquery() + |> join(:inner, [stats], helper in User, on: helper.id == stats.helper_id) + |> join(:left, [stats, _helper], rating in subquery(ratings), + on: rating.reviewee_id == stats.helper_id + ) + |> where(^leaderboard_after(cursor)) + |> order_by( + [stats, helper], + desc: stats.location_supported_people, + desc: stats.verified_people, + desc: stats.unique_people, + desc: stats.completed, + asc: fragment("coalesce(lower(?), '')", helper.display_name), + asc: helper.id + ) + |> limit(^(limit + 1)) + |> select([stats, helper, rating], %{ + helper_id: helper.id, + completed: stats.completed, + unique_people: stats.unique_people, + location_supported_people: stats.location_supported_people, + verified_people: stats.verified_people, + rating: rating.rating, + normalized_name: fragment("coalesce(lower(?), '')", helper.display_name) }) |> Repo.all() end - defp participant?(row, user_id), do: row.helper_id == user_id or row.requester_id == user_id + defp leaderboard_after(nil), do: dynamic(true) - defp counterpart(row, user_id), - do: if(row.helper_id == user_id, do: row.requester_id, else: row.helper_id) - - defp movement_for_user?(row, user_id) do - if row.helper_id == user_id, do: not is_nil(row.helper_movement_observed_at), else: true + defp leaderboard_after(cursor) do + dynamic( + [stats, helper], + stats.location_supported_people < ^cursor.location_supported_people or + (stats.location_supported_people == ^cursor.location_supported_people and + stats.verified_people < ^cursor.verified_people) or + (stats.location_supported_people == ^cursor.location_supported_people and + stats.verified_people == ^cursor.verified_people and + stats.unique_people < ^cursor.unique_people) or + (stats.location_supported_people == ^cursor.location_supported_people and + stats.verified_people == ^cursor.verified_people and + stats.unique_people == ^cursor.unique_people and stats.completed < ^cursor.completed) or + (stats.location_supported_people == ^cursor.location_supported_people and + stats.verified_people == ^cursor.verified_people and + stats.unique_people == ^cursor.unique_people and stats.completed == ^cursor.completed and + fragment("coalesce(lower(?), '')", helper.display_name) > ^cursor.normalized_name) or + (stats.location_supported_people == ^cursor.location_supported_people and + stats.verified_people == ^cursor.verified_people and + stats.unique_people == ^cursor.unique_people and stats.completed == ^cursor.completed and + fragment("coalesce(lower(?), '')", helper.display_name) == ^cursor.normalized_name and + helper.id > ^cursor.helper_id) + ) end - defp ratings(user_id) do - Review - |> where([review], review.reviewee_id == ^user_id and not is_nil(review.revealed_at)) - |> select([review], review.rating) - |> Repo.all() + defp encode_leaderboard_cursor(row) do + %{ + "location_supported_people" => row.location_supported_people, + "verified_people" => row.verified_people, + "unique_people" => row.unique_people, + "completed" => row.completed, + "normalized_name" => row.normalized_name, + "helper_id" => row.helper_id + } + |> Jason.encode!() + |> Base.url_encode64(padding: false) end - defp average([]), do: nil - defp average(values), do: Float.round(Enum.sum(values) / length(values), 1) + defp decode_leaderboard_cursor(value) when value in [nil, ""], do: nil + + defp decode_leaderboard_cursor(value) when is_binary(value) do + with {:ok, json} <- Base.url_decode64(value, padding: false), + {:ok, + %{ + "location_supported_people" => location_supported_people, + "verified_people" => verified_people, + "unique_people" => unique_people, + "completed" => completed, + "normalized_name" => normalized_name, + "helper_id" => helper_id + }} <- Jason.decode(json), + true <- + Enum.all?( + [location_supported_people, verified_people, unique_people, completed], + &(is_integer(&1) and &1 >= 0) + ), + true <- is_binary(normalized_name) and byte_size(normalized_name) <= 80, + {:ok, helper_id} <- Ecto.UUID.cast(helper_id) do + %{ + location_supported_people: location_supported_people, + verified_people: verified_people, + unique_people: unique_people, + completed: completed, + normalized_name: normalized_name, + helper_id: helper_id + } + else + _ -> nil + end + end + + defp decode_leaderboard_cursor(_value), do: nil + defp decimal_average(nil), do: nil defp decimal_average(value), do: value |> Decimal.to_float() |> Float.round(1) + defp before_review(query, nil), do: query + + defp before_review(query, {inserted_at, id}) do + where( + query, + [review], + review.inserted_at < ^inserted_at or + (review.inserted_at == ^inserted_at and review.id < ^id) + ) + end + + defp before_report(query, nil), do: query + + defp before_report(query, {inserted_at, id}) do + where( + query, + [report], + report.inserted_at < ^inserted_at or + (report.inserted_at == ^inserted_at and report.id < ^id) + ) + end + + defp before_block(query, nil), do: query + + defp before_block(query, {inserted_at, id}) do + where( + query, + [block], + block.inserted_at < ^inserted_at or + (block.inserted_at == ^inserted_at and block.id < ^id) + ) + end + + defp before_abuse_signal(query, nil), do: query + + defp before_abuse_signal(query, {inserted_at, id}) do + where( + query, + [signal], + signal.inserted_at < ^inserted_at or + (signal.inserted_at == ^inserted_at and signal.id < ^id) + ) + end + defp authorize_report_target(%Scope{user: user}, %{"request_id" => request_id}) when is_binary(request_id) do case Repo.get(HelpRequest, request_id) do diff --git a/lib/who_need_help_web/live/activity_live/index.ex b/lib/who_need_help_web/live/activity_live/index.ex index ad1a0fb..c9c4e53 100644 --- a/lib/who_need_help_web/live/activity_live/index.ex +++ b/lib/who_need_help_web/live/activity_live/index.ex @@ -25,19 +25,61 @@ defmodule WhoNeedHelpWeb.ActivityLive.Index do {:noreply, socket |> assign(:filters, filters) |> load()} end + def handle_event("load-more-activities", _params, socket) do + page = + Activities.paginate_open_activities( + socket.assigns.current_scope, + socket.assigns.filters, + after: socket.assigns.activities_cursor + ) + + activities = append_unique(socket.assigns.activities, page.entries) + + {:noreply, + socket + |> assign(:activities, activities) + |> assign(:activities_cursor, page.next_cursor) + |> assign(:markers, Jason.encode!(Enum.flat_map(activities, &List.wrap(marker(&1)))))} + end + + def handle_event("load-more-my-activities", _params, socket) do + page = + Activities.paginate_my_activities(socket.assigns.current_scope, + after: socket.assigns.my_activities_cursor + ) + + {:noreply, + socket + |> assign(:my_activities, append_unique(socket.assigns.my_activities, page.entries)) + |> assign(:my_activities_cursor, page.next_cursor)} + end + defp load(socket) do - activities = - Activities.list_open_activities(socket.assigns.current_scope, socket.assigns.filters) + activities_page = + Activities.paginate_open_activities( + socket.assigns.current_scope, + socket.assigns.filters + ) + + my_activities_page = Activities.paginate_my_activities(socket.assigns.current_scope) + activities = activities_page.entries socket |> assign(:page_title, gettext("Activities")) |> assign(:activities, activities) - |> assign(:my_activities, Activities.list_my_activities(socket.assigns.current_scope)) + |> assign(:activities_cursor, activities_page.next_cursor) + |> assign(:my_activities, my_activities_page.entries) + |> assign(:my_activities_cursor, my_activities_page.next_cursor) |> assign(:categories, Catalog.list_categories(:activity)) |> assign(:filter_form, to_form(socket.assigns.filters, as: :filters)) |> assign(:markers, Jason.encode!(Enum.flat_map(activities, &List.wrap(marker(&1))))) end + defp append_unique(existing, incoming) do + existing_ids = MapSet.new(existing, & &1.id) + existing ++ Enum.reject(incoming, &MapSet.member?(existing_ids, &1.id)) + end + defp marker(activity) do case Activity.public_coordinates(activity) do nil -> @@ -151,6 +193,14 @@ defmodule WhoNeedHelpWeb.ActivityLive.Index do + @@ -297,6 +342,14 @@ defmodule WhoNeedHelpWeb.ProfileLive do {gettext("Unblock")} + diff --git a/lib/who_need_help_web/live/request_live/index.ex b/lib/who_need_help_web/live/request_live/index.ex index 1dce4e7..15052aa 100644 --- a/lib/who_need_help_web/live/request_live/index.ex +++ b/lib/who_need_help_web/live/request_live/index.ex @@ -23,20 +23,60 @@ defmodule WhoNeedHelpWeb.RequestLive.Index do {:noreply, socket |> assign(:filters, filters) |> load()} end + def handle_event("load-more-requests", _params, socket) do + page = + Help.paginate_open_requests( + socket.assigns.current_scope, + socket.assigns.filters, + after: socket.assigns.requests_cursor + ) + + requests = append_unique(socket.assigns.requests, page.entries) + + {:noreply, + socket + |> assign(:requests, requests) + |> assign(:requests_cursor, page.next_cursor) + |> assign(:markers, Jason.encode!(Enum.flat_map(requests, &List.wrap(marker(&1)))))} + end + + def handle_event("load-more-my-requests", _params, socket) do + page = + Help.paginate_my_requests(socket.assigns.current_scope, + after: socket.assigns.my_requests_cursor + ) + + {:noreply, + socket + |> assign(:my_requests, append_unique(socket.assigns.my_requests, page.entries)) + |> assign(:my_requests_cursor, page.next_cursor)} + end + defp load(socket) do - requests = Help.list_open_requests(socket.assigns.current_scope, socket.assigns.filters) + requests_page = + Help.paginate_open_requests(socket.assigns.current_scope, socket.assigns.filters) + + my_requests_page = Help.paginate_my_requests(socket.assigns.current_scope) + requests = requests_page.entries user = socket.assigns.current_scope.user socket |> assign(:page_title, gettext("Nearby help")) |> assign(:requests, requests) - |> assign(:my_requests, Help.list_my_requests(socket.assigns.current_scope)) + |> assign(:requests_cursor, requests_page.next_cursor) + |> assign(:my_requests, my_requests_page.entries) + |> assign(:my_requests_cursor, my_requests_page.next_cursor) |> assign(:reputation, Trust.reputation(user.id)) |> assign(:categories, WhoNeedHelp.Catalog.list_categories()) |> assign(:filter_form, to_form(socket.assigns.filters, as: :filters)) |> assign(:markers, Jason.encode!(Enum.flat_map(requests, &List.wrap(marker(&1))))) end + defp append_unique(existing, incoming) do + existing_ids = MapSet.new(existing, & &1.id) + existing ++ Enum.reject(incoming, &MapSet.member?(existing_ids, &1.id)) + end + defp marker(request) do case Help.HelpRequest.public_coordinates(request) do nil -> @@ -178,6 +218,14 @@ defmodule WhoNeedHelpWeb.RequestLive.Index do +