simpletest/who_need_help
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: harden runtime and reduce request overhead

Browse Source
This commit is contained in:
SimpleTest 2026-07-20 06:38:32 +03:00
parent eacfdb2451
commit 16ebd810a5
54 changed files with 1791 additions and 297 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.dialyzer_ignore.exs
View File

@ -11,6 +11,5 @@
# backend path and the same warning class.
[
{"lib/who_need_help/help.ex", :call_without_opaque},
{"lib/who_need_help/tracking.ex", :call_without_opaque},
{"lib/who_need_help_web/gettext.ex", :call_without_opaque}
]

7
.env.e2e.example
View File

@ -3,6 +3,7 @@
# reuses the staging Compose project or database volume.
HTTP_PORT=0
MAILPIT_PORT=0
DOCKER_SOCKET_GID=REPLACE_WITH_DOCKER_SOCKET_NUMERIC_GID
TRAEFIK_TRUSTED_IPS=127.0.0.1/32
TRAEFIK_RETRY_ATTEMPTS=3
TRAEFIK_PROJECT_CONSTRAINT=GENERATED_UNIQUE_E2E_PROJECT
@ -19,7 +20,11 @@ POSTGRES_DB=who_need_help_e2e
POSTGRES_USER=postgres
POSTGRES_PASSWORD=GENERATE_INDEPENDENT_E2E_DATABASE_PASSWORD
DATABASE_URL=ecto://postgres:GENERATE_URL_SAFE_PASSWORD@db/who_need_help_e2e
POOL_SIZE=10
WEB_POOL_SIZE=4
WORKER_POOL_SIZE=2
MIGRATE_POOL_SIZE=2
WEB_REPLICAS=2
WORKER_REPLICAS=2
ERLANG_PORT_LIMIT=65536
SECRET_KEY_BASE=GENERATE_INDEPENDENT_E2E_SECRET_KEY_BASE

10
.env.example
View File

@ -2,6 +2,8 @@
# required values. Replace every credential before any public deployment.
HTTP_PORT=4010
MAILPIT_PORT=8027
MAILPIT_BIND_ADDRESS=127.0.0.1
DOCKER_SOCKET_GID=REPLACE_WITH_DOCKER_SOCKET_NUMERIC_GID
# Comma-separated proxy IP/CIDR values whose X-Forwarded-* headers Traefik
# accepts. Keep loopback locally; set the exact VPN proxy address for staging.
TRAEFIK_TRUSTED_IPS=127.0.0.1/32
@ -10,7 +12,7 @@ TRAEFIK_RETRY_ATTEMPTS=3
# own project constraint, router/service name, Docker network, and Host rule.
TRAEFIK_PROJECT_CONSTRAINT=who_need_help
TRAEFIK_APP_NAME=who-need-help
TRAEFIK_DOCKER_NETWORK=who_need_help_internal
TRAEFIK_DOCKER_NETWORK=who_need_help_ingress
TRAEFIK_ROUTER_RULE='PathPrefix(`/`)'
PHX_HOST=localhost
PHX_SCHEME=http
@ -62,7 +64,11 @@ POSTGRES_USER=postgres
POSTGRES_PASSWORD=replace-with-a-local-or-deployment-secret
DATABASE_URL=ecto://postgres:replace-with-url-encoded-password@db/who_need_help
POOL_SIZE=10
WEB_POOL_SIZE=4
WORKER_POOL_SIZE=2
MIGRATE_POOL_SIZE=2
WEB_REPLICAS=2
WORKER_REPLICAS=2
# Maximum simultaneously existing Erlang ports (files, sockets and drivers).
# Keeping this explicit prevents a host's very large nofile ulimit from making
# every BEAM instance preallocate a multi-gigabyte port table.

9
.env.load.example
View File

@ -6,6 +6,7 @@ LOAD_PROJECT=who_need_help_load
LOAD_HOST=load.local
HTTP_PORT=4110
MAILPIT_PORT=8127
DOCKER_SOCKET_GID=REPLACE_WITH_DOCKER_SOCKET_NUMERIC_GID
PHX_HOST=load.local
PHX_SCHEME=https
PHX_URL_PORT=443
@ -13,14 +14,18 @@ TRAEFIK_TRUSTED_IPS=127.0.0.1/32
TRAEFIK_RETRY_ATTEMPTS=3
TRAEFIK_PROJECT_CONSTRAINT=who_need_help_load
TRAEFIK_APP_NAME=who-need-help-load
TRAEFIK_DOCKER_NETWORK=who_need_help_load_internal
TRAEFIK_DOCKER_NETWORK=who_need_help_load_ingress
TRAEFIK_ROUTER_RULE='Host(`load.local`)'
POSTGRES_DB=who_need_help_load
POSTGRES_USER=wnh_load
POSTGRES_PASSWORD=GENERATE_POSTGRES_PASSWORD
DATABASE_URL=GENERATE_DATABASE_URL
POOL_SIZE=10
WEB_POOL_SIZE=4
WORKER_POOL_SIZE=2
MIGRATE_POOL_SIZE=2
WEB_REPLICAS=2
WORKER_REPLICAS=2
ERLANG_PORT_LIMIT=65536
SECRET_KEY_BASE=GENERATE_SECRET_KEY_BASE
HANDOVER_SECRET=GENERATE_HANDOVER_SECRET

9
Dockerfile.postgis Normal file
View File

@ -0,0 +1,9 @@
# syntax=docker/dockerfile:1.20.0
FROM postgis/postgis:18-3.6-alpine@sha256:05d68c7f0f19b9aa0bf7c4a2049b2e8b38b44a63116392b95726a4c913766cf6
# PostgreSQL 18's image-owned volume path is already writable by this account.
# Starting non-root removes the entrypoint's need for the bundled gosu binary.
RUN rm /usr/local/bin/gosu
USER postgres

11
Dockerfile.socket-proxy Normal file
View File

@ -0,0 +1,11 @@
# syntax=docker/dockerfile:1.20.0
FROM tecnativa/docker-socket-proxy:v0.4.2@sha256:1f3a6f303320723d199d2316a3e82b2e2685d86c275d5e3deeaf182573b47476
USER root
# The upstream release image is immutable but its Alpine packages predate
# currently available security fixes. Keep the reviewed proxy implementation
# and apply the repository's current fixes; scripts/quality.sh scans the result.
RUN apk upgrade --no-cache
USER haproxy

14
compose.cpu-replay.yaml Normal file
View File

@ -0,0 +1,14 @@
# Optional measurement-only override. It constrains CPU visibility for each
# long-running BEAM container so scheduler-count and memory behavior can be
# reproduced on a many-core development host. Values are experiment inputs,
# not production recommendations or minimum requirements.
services:
web:
cpus: ${CPU_REPLAY_WEB_CPUS:?Set CPU_REPLAY_WEB_CPUS for this experiment}
environment:
ERL_ZFLAGS: "+Q ${ERLANG_PORT_LIMIT:-65536} +S ${CPU_REPLAY_WEB_SCHEDULERS:?Set CPU_REPLAY_WEB_SCHEDULERS}:${CPU_REPLAY_WEB_SCHEDULERS:?Set CPU_REPLAY_WEB_SCHEDULERS}"
worker:
cpus: ${CPU_REPLAY_WORKER_CPUS:?Set CPU_REPLAY_WORKER_CPUS for this experiment}
environment:
ERL_ZFLAGS: "+Q ${ERLANG_PORT_LIMIT:-65536} +S ${CPU_REPLAY_WORKER_SCHEDULERS:?Set CPU_REPLAY_WORKER_SCHEDULERS}:${CPU_REPLAY_WORKER_SCHEDULERS:?Set CPU_REPLAY_WORKER_SCHEDULERS}"

2
compose.load.yaml
View File

@ -15,7 +15,7 @@ services:
edge:
aliases:
- ${LOAD_HOST}
internal: {}
ingress: {}
migrate:
image: who-need-help:load

2
compose.quality.yaml
View File

@ -1,6 +1,6 @@
services:
db:
image: postgis/postgis:18-3.6-alpine@sha256:05d68c7f0f19b9aa0bf7c4a2049b2e8b38b44a63116392b95726a4c913766cf6
image: ${QUALITY_POSTGIS_IMAGE:?Set QUALITY_POSTGIS_IMAGE}
environment:
POSTGRES_DB: postgres
POSTGRES_USER: ${QUALITY_POSTGRES_USER:?Set QUALITY_POSTGRES_USER}

99
compose.yaml
View File

@ -16,7 +16,6 @@ x-app-environment: &app-environment
PHX_SCHEME: ${PHX_SCHEME:?Set PHX_SCHEME in .env}
PHX_URL_PORT: ${PHX_URL_PORT:?Set PHX_URL_PORT in .env}
PORT: "4000"
POOL_SIZE: ${POOL_SIZE:?Set POOL_SIZE in .env after measuring the target profile}
SMTP_RELAY: ${SMTP_RELAY:?Set SMTP_RELAY in .env}
SMTP_PORT: ${SMTP_PORT:?Set SMTP_PORT in .env}
SMTP_USERNAME: ${SMTP_USERNAME:-}
@ -45,11 +44,40 @@ x-app-environment: &app-environment
PUSH_HTTP_RETRY_DELAY_MS: ${PUSH_HTTP_RETRY_DELAY_MS:-}
services:
docker-api-proxy:
image: who-need-help:socket-proxy-local
build:
context: .
dockerfile: Dockerfile.socket-proxy
environment:
CONTAINERS: "1"
EVENTS: "1"
INFO: "1"
NETWORKS: "1"
PING: "1"
POST: "0"
VERSION: "1"
LOG_LEVEL: warning
volumes:
- /var/run/docker.sock:/var/run/docker.sock:ro
group_add:
- ${DOCKER_SOCKET_GID:?Set DOCKER_SOCKET_GID to the numeric group of /var/run/docker.sock}
networks: [docker-api]
read_only: true
tmpfs:
- /run:uid=99,gid=99,mode=0755
- /tmp
cap_drop: [ALL]
security_opt:
- no-new-privileges:true
restart: unless-stopped
proxy:
image: traefik:v3.7.8@sha256:4299bbed850421258fc5448c2e0e6ad350981d4d335a68de11b92448aedbefe5
command:
- --api.dashboard=false
- --providers.docker=true
- --providers.docker.endpoint=tcp://docker-api-proxy:2375
- --providers.docker.exposedbydefault=false
- --providers.docker.constraints=Label(`com.docker.compose.project`,`${TRAEFIK_PROJECT_CONSTRAINT:-who_need_help}`)
- --entrypoints.web.address=:80
@ -57,13 +85,22 @@ services:
- --entrypoints.web.forwardedheaders.trustedips=${TRAEFIK_TRUSTED_IPS:-127.0.0.1/32}
ports:
- "${HTTP_PORT:-4010}:80"
volumes:
- /var/run/docker.sock:/var/run/docker.sock:ro
networks: [edge, internal]
depends_on:
- docker-api-proxy
networks: [docker-api, edge, ingress]
read_only: true
tmpfs:
- /tmp
cap_drop: [ALL]
security_opt:
- no-new-privileges:true
restart: unless-stopped
db:
image: postgis/postgis:18-3.6-alpine@sha256:05d68c7f0f19b9aa0bf7c4a2049b2e8b38b44a63116392b95726a4c913766cf6
image: who-need-help:postgis-local
build:
context: .
dockerfile: Dockerfile.postgis
environment:
POSTGRES_DB: ${POSTGRES_DB:?Set POSTGRES_DB in .env}
POSTGRES_USER: ${POSTGRES_USER:?Set POSTGRES_USER in .env}
@ -79,13 +116,22 @@ services:
volumes:
- postgres_data:/var/lib/postgresql
networks: [internal]
security_opt:
- no-new-privileges:true
restart: unless-stopped
mailpit:
image: axllent/mailpit:v1.30.4@sha256:5a49a77c5bdbe7c5474450b4f46348d09949df3695257729c93a30369382d4f6
user: "65534:65534"
ports:
- "${MAILPIT_PORT:-8027}:8025"
- "${MAILPIT_BIND_ADDRESS:-127.0.0.1}:${MAILPIT_PORT:-8027}:8025"
networks: [edge, internal]
read_only: true
tmpfs:
- /tmp
cap_drop: [ALL]
security_opt:
- no-new-privileges:true
restart: unless-stopped
migrate:
@ -95,11 +141,18 @@ services:
environment:
<<: *app-environment
APP_ROLE: migrate
POOL_SIZE: ${MIGRATE_POOL_SIZE:-${POOL_SIZE:?Set MIGRATE_POOL_SIZE in .env}}
command: ["/app/bin/migrate"]
depends_on:
db:
condition: service_healthy
networks: [internal]
read_only: true
tmpfs:
- /tmp
cap_drop: [ALL]
security_opt:
- no-new-privileges:true
restart: "no"
web:
@ -108,12 +161,13 @@ services:
<<: *app-environment
APP_ROLE: web
PHX_SERVER: "true"
POOL_SIZE: ${WEB_POOL_SIZE:-${POOL_SIZE:?Set WEB_POOL_SIZE in .env}}
depends_on:
migrate:
condition: service_completed_successfully
labels:
- traefik.enable=true
- traefik.docker.network=${TRAEFIK_DOCKER_NETWORK:-who_need_help_internal}
- traefik.docker.network=${TRAEFIK_DOCKER_NETWORK:-who_need_help_ingress}
- traefik.http.routers.${TRAEFIK_APP_NAME:-who-need-help}.rule=${TRAEFIK_ROUTER_RULE:-PathPrefix(`/`)}
- traefik.http.routers.${TRAEFIK_APP_NAME:-who-need-help}.entrypoints=web
- traefik.http.routers.${TRAEFIK_APP_NAME:-who-need-help}.service=${TRAEFIK_APP_NAME:-who-need-help}
@ -126,8 +180,14 @@ services:
timeout: 3s
retries: 10
deploy:
replicas: 2
networks: [internal]
replicas: ${WEB_REPLICAS:-2}
networks: [ingress, internal, egress]
read_only: true
tmpfs:
- /tmp
cap_drop: [ALL]
security_opt:
- no-new-privileges:true
restart: unless-stopped
worker:
@ -135,19 +195,36 @@ services:
environment:
<<: *app-environment
APP_ROLE: worker
POOL_SIZE: ${WORKER_POOL_SIZE:-${POOL_SIZE:?Set WORKER_POOL_SIZE in .env}}
command: ["/app/bin/who_need_help", "start"]
depends_on:
migrate:
condition: service_completed_successfully
healthcheck:
test: ["CMD", "curl", "--fail", "--silent", "http://localhost:4000/healthz/ready"]
interval: 10s
timeout: 3s
retries: 10
deploy:
replicas: 2
networks: [internal]
replicas: ${WORKER_REPLICAS:-2}
networks: [internal, egress]
read_only: true
tmpfs:
- /tmp
cap_drop: [ALL]
security_opt:
- no-new-privileges:true
restart: unless-stopped
networks:
docker-api:
internal: true
edge:
ingress:
internal: true
internal:
internal: true
egress:
volumes:
postgres_data:

6
config/runtime.exs
View File

@ -179,16 +179,16 @@ config :who_need_help,
),
push_delivery_options: Keyword.delete(push_configuration, :adapter)
if config_env() == :prod and app_role == :web do
if config_env() == :prod and app_role in [:web, :worker] do
metrics_token =
System.get_env("METRICS_TOKEN") ||
raise """
environment variable METRICS_TOKEN is missing for the web role.
environment variable METRICS_TOKEN is missing for the web or worker role.
Generate an independent random value and store it in the deployment secret.
"""
if metrics_token == "" do
raise "METRICS_TOKEN must not be empty for the web role."
raise "METRICS_TOKEN must not be empty for the web or worker role."
end
config :who_need_help, :metrics_token, metrics_token

4
deploy/helm/who-need-help/templates/deployments.yaml
View File

@ -80,7 +80,7 @@ spec:
- name: PORT
value: {{ $root.Values.app.port | quote }}
- name: POOL_SIZE
value: {{ $root.Values.app.poolSize | quote }}
value: {{ $settings.poolSize | quote }}
- name: SMTP_RELAY
value: {{ $root.Values.app.smtpRelay | quote }}
- name: SMTP_PORT
@ -93,7 +93,6 @@ spec:
value: {{ $root.Values.app.mapTileUrl | quote }}
- name: DNS_CLUSTER_QUERY
value: "{{ include "who-need-help.fullname" $root }}-headless.{{ $root.Release.Namespace }}.svc.cluster.local"
{{- if eq $component "web" }}
ports:
- name: http
containerPort: 4000
@ -107,7 +106,6 @@ spec:
path: /healthz/live
port: http
periodSeconds: 10
{{- end }}
securityContext:
allowPrivilegeEscalation: false
readOnlyRootFilesystem: true

2
deploy/helm/who-need-help/templates/migrate-job.yaml
View File

@ -35,6 +35,8 @@ spec:
value: migrate
- name: ERL_ZFLAGS
value: {{ printf "+Q %d" (int .Values.app.erlangPortLimit) | quote }}
- name: POOL_SIZE
value: {{ .Values.app.migratePoolSize | quote }}
securityContext:
allowPrivilegeEscalation: false
readOnlyRootFilesystem: true

17
deploy/helm/who-need-help/templates/services.yaml
View File

@ -19,6 +19,23 @@ spec:
---
apiVersion: v1
kind: Service
metadata:
name: {{ include "who-need-help.fullname" . }}-worker-metrics
labels:
{{- include "who-need-help.labels" . | nindent 4 }}
app.kubernetes.io/component: worker
spec:
type: ClusterIP
selector:
{{- include "who-need-help.selectorLabels" . | nindent 4 }}
app.kubernetes.io/component: worker
ports:
- name: http
port: 4000
targetPort: http
---
apiVersion: v1
kind: Service
metadata:
name: {{ include "who-need-help.fullname" . }}-headless
labels:

4
deploy/helm/who-need-help/values.yaml
View File

@ -5,9 +5,11 @@ image:
web:
replicas: 2
poolSize: "4"
worker:
replicas: 2
poolSize: "2"
service:
type: ClusterIP
@ -19,7 +21,7 @@ app:
scheme: https
urlPort: "443"
port: "4000"
poolSize: "10"
migratePoolSize: "2"
# OTP otherwise derives this from the container nofile ulimit. Some
# Kubernetes runtimes expose a value so large that each BEAM instance
# preallocates a multi-gigabyte port table.

46
docs/performance.md
View File

@ -21,6 +21,23 @@ before any rollout. Compose and Helm now set that value explicitly, and the
startup and rolling-verification scripts check the effective value in every
running web and worker VM.
On a host with more CPUs than the intended server, the optional
`compose.cpu-replay.yaml` override can constrain the CPUs visible to each
long-running BEAM container:
```sh
CPU_REPLAY_WEB_CPUS=1 CPU_REPLAY_WORKER_CPUS=1 \
CPU_REPLAY_WEB_SCHEDULERS=1 CPU_REPLAY_WORKER_SCHEDULERS=1 \
docker compose -f compose.yaml -f compose.cpu-replay.yaml up -d --wait
```
All four values are explicit experiment inputs. The override does not set a memory
limit or represent a whole-server CPU budget. The scheduler inputs make total
and online BEAM scheduler counts match the intended replay instead of leaving
offline scheduler threads sized from the development host. None of these
inputs is a production sizing recommendation. Running ordinary
`docker compose up` without the override removes the replay constraints.
After the Helm rollout, the unchanged kind control-plane container measured
1.796 GiB, a reduction of 7.952 GiB (81.6%). Its four application pod cgroups
measured 188.5–215.6 MiB in the first post-change sample. A subsequent
@ -295,6 +312,35 @@ Ignored before/after evidence:
- `output/performance/pool-sql-final-20260720/`
- `output/performance/pool-sql-chat-fixed-canonical-20260720/`
## Observed role-specific connection pools
On 2026-07-20 the same 3-web/2-worker, 40 public HTTP VU, 40 Phoenix
heartbeat VU, and 8 authenticated mutual-aid VU profile was repeated with
four Repo connections per web replica and two per worker replica. The migrate
role also started and completed with a two-connection pool. These are measured
development defaults, not production minimums.
The 16 configured long-running Repo slots produced 38,678 HTTP requests with
zero failures, 1,760 complete authenticated page/chat/tracking chains, and 240
Phoenix heartbeat sessions. The application recorded 131,409 web-side queries
with 34.82 microseconds average pool queue time, compared with 31.15
microseconds in the earlier 10-connections-per-replica run. PostgreSQL observed
21 peak client backends, 6 peak active backends, 76 minimum non-reserved
connection headroom, no idle transaction, no lock wait, no rollback, no
deadlock, no conflict, and no temporary file.
The run-scoped application log had no match for DBConnection/Postgrex
connection errors, pool timeouts, exceptions, deadlocks, `FATAL`, or `PANIC`.
All 1,760 messages, tracking sessions, and tracking samples were recorded;
explicit stop left zero current positions, and fixture cleanup restored the
tracked table counts exactly. The result supports using web=4, worker=2, and
migrate=2 as the repository's current measured starting profile. It does not
establish saturation, an SLO, or a production capacity limit.
Ignored evidence:
- `output/performance/role-pools-4-2-20260720/`
## Observed 10-minute authenticated soak
Observed on 2026-07-19 with the same 3-web/2-worker topology and the same

67
lib/who_need_help/accounts.ex
View File

@ -11,6 +11,43 @@ defmodule WhoNeedHelp.Accounts do
## Database getters
@public_user_fields [:id, :display_name, :bio, :tip_url, :inserted_at]
@public_social_identity_fields [
:id,
:provider,
:profile_url,
:handle,
:verified_at,
:user_id,
:inserted_at
]
@doc """
Returns the deliberately small user projection used by public and
participant-facing views.
Authentication, account settings, and moderation queries must continue to
load the full schema explicitly. In particular, this projection never loads
email, password hashes, roles, or moderation notes.
"""
def public_user_query(options \\ []) do
query =
from user in User,
select: struct(user, ^@public_user_fields)
if Keyword.get(options, :social_identities, false) do
public_social_identity = public_social_identity_query()
preload(query, social_identities: ^public_social_identity)
else
query
end
end
def public_social_identity_query do
from identity in SocialIdentity,
select: struct(identity, ^@public_social_identity_fields)
end
@doc """
Gets a user by email.
@ -72,33 +109,31 @@ defmodule WhoNeedHelp.Accounts do
def eligible_for_trust_actions?(_user), do: false
def eligible_user_id?(user_id) do
case Repo.get(User, user_id) do
%User{} = user -> eligible_for_trust_actions?(user)
nil -> false
end
Repo.exists?(
from user in User,
where:
user.id == ^user_id and user.moderation_status == :active and
not is_nil(user.confirmed_at) and not is_nil(user.accepted_terms_at)
)
end
def moderator?(%User{role: role}), do: role in [:moderator, :admin]
def moderator?(_user), do: false
def moderator_authorized?(%User{id: id}) do
case Repo.get(User, id) do
%User{} = user -> moderator?(user)
nil -> false
end
end
def moderator_authorized?(%User{id: id, role: role})
when role in [:moderator, :admin],
do:
Repo.exists?(
from user in User, where: user.id == ^id and user.role in [:moderator, :admin]
)
def moderator_authorized?(_user), do: false
def admin?(%User{role: :admin}), do: true
def admin?(_user), do: false
def admin_authorized?(%User{id: id}) do
case Repo.get(User, id) do
%User{} = user -> admin?(user)
nil -> false
end
end
def admin_authorized?(%User{id: id, role: :admin}),
do: Repo.exists?(from user in User, where: user.id == ^id and user.role == :admin)
def admin_authorized?(_user), do: false

72
lib/who_need_help/activities.ex
View File

@ -5,6 +5,7 @@ defmodule WhoNeedHelp.Activities do
import Ecto.Query
alias WhoNeedHelp.Accounts
alias WhoNeedHelp.Accounts.Scope
alias WhoNeedHelp.Activities.{Activity, Message, Participant}
alias WhoNeedHelp.Catalog
@ -36,6 +37,7 @@ defmodule WhoNeedHelp.Activities do
now = DateTime.utc_now(:second)
limit = Pagination.limit(options)
cursor = Pagination.cursor(options)
public_creator = Accounts.public_user_query(social_identities: true)
Activity
|> where(
@ -59,9 +61,9 @@ defmodule WhoNeedHelp.Activities do
|> 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)
|> with_approved_participant_count()
|> preload([activity], category: :parent, creator: ^public_creator)
|> Repo.all()
|> Enum.map(&public_activity/1)
|> Pagination.page(limit, &{&1.starts_at, &1.id})
end
@ -82,7 +84,8 @@ defmodule WhoNeedHelp.Activities do
|> before_my_activity(cursor)
|> order_by([activity], desc: activity.starts_at, desc: activity.id)
|> limit(^(limit + 1))
|> preload(category: :parent, participants: :user)
|> with_approved_participant_count()
|> preload([activity], category: :parent)
|> Repo.all()
|> Pagination.page(limit, &{&1.starts_at, &1.id})
end
@ -121,13 +124,11 @@ defmodule WhoNeedHelp.Activities do
end
def get_activity(%Scope{user: user}, id) do
case Repo.get(Activity, id) do
case get_loaded_activity(id) do
nil ->
{:error, :not_found}
activity ->
activity = load_activity(activity.id)
cond do
activity.creator_id == user.id ->
{:ok, activity}
@ -154,6 +155,7 @@ defmodule WhoNeedHelp.Activities do
if activity.creator_id == user.id or approved_participant?(activity, user.id) do
limit = Pagination.limit(options, 50)
cursor = Pagination.cursor(options)
public_user = Accounts.public_user_query()
page =
Message
@ -161,7 +163,7 @@ defmodule WhoNeedHelp.Activities do
|> before_message(cursor)
|> order_by([message], desc: message.inserted_at, desc: message.id)
|> limit(^(limit + 1))
|> preload(:sender)
|> preload([message], sender: ^public_user)
|> Repo.all()
|> Pagination.page(limit, &{&1.inserted_at, &1.id})
@ -356,7 +358,7 @@ defmodule WhoNeedHelp.Activities do
|> Repo.insert()
|> case do
{:ok, message} ->
message = Repo.preload(message, :sender)
message = Repo.preload(message, sender: Accounts.public_user_query())
Phoenix.PubSub.broadcast(
WhoNeedHelp.PubSub,
@ -559,18 +561,34 @@ 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(
|> preload_activity()
end
defp get_loaded_activity(id) do
Activity
|> Repo.get(id)
|> preload_activity()
end
defp preload_activity(nil), do: nil
defp preload_activity(activity) do
public_user = Accounts.public_user_query(social_identities: true)
public_sender = Accounts.public_user_query()
recent_messages =
Message
|> order_by([message], desc: message.inserted_at, desc: message.id)
|> limit(50)
|> preload([message], sender: ^public_sender)
Repo.preload(
activity,
category: :parent,
creator: :social_identities,
participants: [user: :social_identities],
creator: public_user,
participants: [user: public_user],
messages: recent_messages
)
end
@ -587,8 +605,24 @@ defmodule WhoNeedHelp.Activities do
}
end
defp public_activity(activity) do
%{activity | participants: Enum.filter(activity.participants, &(&1.status == :approved))}
defp with_approved_participant_count(query) do
counts =
Participant
|> where([participant], participant.status == :approved)
|> group_by([participant], participant.activity_id)
|> select([participant], %{
activity_id: participant.activity_id,
approved_count: count(participant.id)
})
query
|> join(:left, [activity], count in subquery(counts),
on: count.activity_id == activity.id,
as: :approved_counts
)
|> select_merge([approved_counts: count], %{
approved_participant_count: fragment("coalesce(?, 0)", count.approved_count)
})
end
defp locked_activity(id) do

1
lib/who_need_help/activities/activity.ex
View File

@ -24,6 +24,7 @@ defmodule WhoNeedHelp.Activities.Activity do
field :completed_at, :utc_datetime
field :hidden_at, :utc_datetime
field :hidden_reason, :string
field :approved_participant_count, :integer, virtual: true, default: 0
belongs_to :creator, WhoNeedHelp.Accounts.User
belongs_to :category, WhoNeedHelp.Catalog.Category

16
lib/who_need_help/application.ex
View File

@ -26,9 +26,19 @@ defmodule WhoNeedHelp.Application do
role_children =
case Application.fetch_env!(:who_need_help, :app_role) do
:web -> [{Oban, oban_client_config}, WhoNeedHelpWeb.Presence, WhoNeedHelpWeb.Endpoint]
:worker -> [{Oban, oban_config}]
:migrate -> [{Oban, oban_client_config}]
:web ->
[
{Oban, oban_client_config},
WhoNeedHelpWeb.Presence,
WhoNeedHelp.TrackingPresenceCleanup,
WhoNeedHelpWeb.Endpoint
]
:worker ->
[{Oban, oban_config}, WhoNeedHelpWeb.WorkerMetricsPlug]
:migrate ->
[{Oban, oban_client_config}]
end
children = common_children ++ role_children

23
lib/who_need_help/catalog.ex
View File

@ -40,13 +40,15 @@ defmodule WhoNeedHelp.Catalog do
def paginate_proposals(options \\ []) do
limit = Pagination.limit(options)
cursor = Pagination.cursor(options)
public_user = Accounts.public_user_query()
CategoryProposal
|> where([p], p.status == :open)
|> preload([:proposer, :parent, :votes])
|> before_proposal(cursor)
|> order_by([proposal], desc: proposal.inserted_at, desc: proposal.id)
|> limit(^(limit + 1))
|> with_vote_count()
|> preload([proposal], proposer: ^public_user, parent: [])
|> Repo.all()
|> Pagination.page(limit, &{&1.inserted_at, &1.id})
end
@ -163,7 +165,8 @@ defmodule WhoNeedHelp.Catalog do
|> before_proposal(cursor)
|> order_by([proposal], desc: proposal.inserted_at, desc: proposal.id)
|> limit(^(limit + 1))
|> preload([:proposer, :parent, :merged_into, :reviewed_by, :votes])
|> with_vote_count()
|> preload([proposal], [:parent, :merged_into])
|> Repo.all()
|> Pagination.page(limit, &{&1.inserted_at, &1.id})
else
@ -833,6 +836,22 @@ defmodule WhoNeedHelp.Catalog do
]
end
defp with_vote_count(query) do
counts =
CategoryVote
|> group_by([vote], vote.proposal_id)
|> select([vote], %{proposal_id: vote.proposal_id, vote_count: count(vote.id)})
query
|> join(:left, [proposal], count in subquery(counts),
on: count.proposal_id == proposal.id,
as: :vote_counts
)
|> select_merge([vote_counts: count], %{
vote_count: fragment("coalesce(?, 0)", count.vote_count)
})
end
defp before_proposal(query, nil), do: query
defp before_proposal(query, {inserted_at, id}) do

1
lib/who_need_help/catalog/category_proposal.ex
View File

@ -16,6 +16,7 @@ defmodule WhoNeedHelp.Catalog.CategoryProposal do
belongs_to :reviewed_by, WhoNeedHelp.Accounts.User
field :reviewed_at, :utc_datetime
field :moderation_note, :string
field :vote_count, :integer, virtual: true, default: 0
has_many :votes, WhoNeedHelp.Catalog.CategoryVote, foreign_key: :proposal_id
timestamps(type: :utc_datetime)

2
lib/who_need_help/catalog_moderation.ex
View File

@ -19,7 +19,7 @@ defmodule WhoNeedHelp.CatalogModeration do
reason: proposal.reason,
mode: proposal.mode,
parent_slug: parent_slug(proposal.parent),
community_votes: length(proposal.votes)
community_votes: proposal.vote_count
}
end)

103
lib/who_need_help/help.ex
View File

@ -57,8 +57,8 @@ defmodule WhoNeedHelp.Help do
|> after_open_request(cursor)
|> order_by([request], asc: request.expires_at, asc: request.id)
|> limit(^(limit + 1))
|> preload([:category, :requester, assignment: :helper])
|> Repo.all()
|> preload_request_relations()
|> Pagination.page(limit, &{&1.expires_at, &1.id})
end
@ -77,8 +77,8 @@ defmodule WhoNeedHelp.Help do
|> before_my_request(cursor)
|> order_by([request], desc: request.inserted_at, desc: request.id)
|> limit(^(limit + 1))
|> preload([:category, :requester, assignment: :helper])
|> Repo.all()
|> preload_request_relations()
|> Pagination.page(limit, &{&1.inserted_at, &1.id})
end
@ -107,31 +107,39 @@ defmodule WhoNeedHelp.Help do
def get_request!(id) do
HelpRequest
|> Repo.get!(id)
|> Repo.preload([
:category,
requester: :social_identities,
assignment: [helper: :social_identities]
])
|> then(&preload_request_relations([&1], social_identities: true))
|> hd()
end
def get_request(%Scope{user: user} = scope, id) do
case Repo.get(HelpRequest, id) do
request =
HelpRequest
|> Repo.get(id)
|> case do
nil ->
nil
request ->
[request]
|> preload_request_relations(social_identities: true)
|> hd()
end
case request do
nil ->
{:error, :not_found}
request ->
request = get_request!(request.id)
cond do
Accounts.moderator_authorized?(user) ->
{:ok, request}
request.requester_id == user.id ->
{:ok, request}
request.assignment && participant?(scope, request.assignment) ->
{:ok, request}
Accounts.moderator_authorized?(user) ->
{:ok, request}
not is_nil(request.hidden_at) ->
{:error, :not_found}
@ -146,7 +154,10 @@ defmodule WhoNeedHelp.Help do
def get_assignment_for_participant(%Scope{} = scope, id) do
with {:ok, id} <- Ecto.UUID.cast(id),
%Assignment{} = assignment <- Repo.get(Assignment, id),
%Assignment{} = assignment <-
Assignment
|> Repo.get(id)
|> Repo.preload(:request),
true <- participant?(scope, assignment) do
{:ok, assignment}
else
@ -248,7 +259,12 @@ defmodule WhoNeedHelp.Help do
{:ok, assignment} ->
request = get_request!(assignment.request_id)
broadcast({:request_updated, request})
{:ok, Repo.preload(assignment, [:helper, :request])}
{:ok,
Repo.preload(assignment,
helper: Accounts.public_user_query(),
request: []
)}
other ->
other
@ -389,6 +405,55 @@ defmodule WhoNeedHelp.Help do
user.id in [assignment.helper_id, request.requester_id]
end
defp preload_request_relations(requests, options \\ []) do
requests = Repo.preload(requests, [:category, :assignment])
user_ids =
requests
|> Enum.flat_map(fn request ->
[request.requester_id, request.assignment && request.assignment.helper_id]
end)
|> Enum.reject(&is_nil/1)
|> Enum.uniq()
users =
Accounts.public_user_query()
|> where([user], user.id in ^user_ids)
|> Repo.all()
|> maybe_preload_social_identities(options)
|> Map.new(&{&1.id, &1})
Enum.map(requests, fn request ->
assignment =
case request.assignment do
%Assignment{} = assignment ->
%{assignment | helper: Map.fetch!(users, assignment.helper_id)}
nil ->
nil
end
request
|> Map.put(:requester, Map.fetch!(users, request.requester_id))
|> Map.put(:assignment, assignment)
|> attach_assignment_request()
end)
end
defp maybe_preload_social_identities(users, options) do
if Keyword.get(options, :social_identities, false) do
Repo.preload(users, social_identities: Accounts.public_social_identity_query())
else
users
end
end
defp attach_assignment_request(%HelpRequest{assignment: %Assignment{} = assignment} = request) do
%{request | assignment: %{assignment | request: request}}
end
defp attach_assignment_request(%HelpRequest{} = request), do: request
def requester?(%Scope{user: user}, %HelpRequest{requester_id: id}), do: user.id == id
def helper?(%Scope{user: user}, %Assignment{helper_id: id}), do: user.id == id
@ -506,7 +571,13 @@ defmodule WhoNeedHelp.Help do
request = get_request!(assignment.request_id)
broadcast({:request_updated, request})
if assignment.status == :completed, do: Trust.record_completion_signals(assignment)
{:ok, Repo.preload(assignment, [:helper, :request], force: true)}
{:ok,
Repo.preload(
assignment,
[helper: Accounts.public_user_query(), request: []],
force: true
)}
end
defp after_transition(other), do: other

12
lib/who_need_help/messaging.ex
View File

@ -2,6 +2,7 @@ defmodule WhoNeedHelp.Messaging do
@moduledoc "Durable match chat with PubSub fan-out after commit."
import Ecto.Query
alias WhoNeedHelp.Accounts
alias WhoNeedHelp.Accounts.Scope
alias WhoNeedHelp.Help
alias WhoNeedHelp.Help.Assignment
@ -20,10 +21,13 @@ defmodule WhoNeedHelp.Messaging do
end
def paginate_messages(%Scope{} = scope, %Assignment{} = assignment, options \\ []) do
assignment = Repo.preload(assignment, :request)
if Trust.eligible?(scope) and Help.participant?(scope, assignment) and
not blocked_assignment?(scope, assignment) do
limit = Pagination.limit(options, 50)
cursor = Pagination.cursor(options)
public_user = Accounts.public_user_query()
page =
Message
@ -31,7 +35,7 @@ defmodule WhoNeedHelp.Messaging do
|> before_message(cursor)
|> order_by([message], desc: message.inserted_at, desc: message.id)
|> limit(^(limit + 1))
|> preload(:sender)
|> preload([message], sender: ^public_user)
|> Repo.all()
|> Pagination.page(limit, &{&1.inserted_at, &1.id})
@ -42,10 +46,12 @@ defmodule WhoNeedHelp.Messaging do
end
def send_message(%Scope{user: user} = scope, %Assignment{} = assignment, attrs) do
assignment = Repo.preload(assignment, :request)
with {:ok, _limit} <- Trust.authorize_action(scope, :send_message),
true <- Help.participant?(scope, assignment),
false <- blocked_assignment?(scope, assignment) do
request = assignment |> Repo.preload(:request) |> Map.fetch!(:request)
request = Map.fetch!(assignment, :request)
recipient_id = counterpart_id(user.id, assignment, request)
result =
@ -66,7 +72,7 @@ defmodule WhoNeedHelp.Messaging do
end)
with {:ok, message} <- result do
message = Repo.preload(message, :sender)
message = Repo.preload(message, sender: Accounts.public_user_query())
Phoenix.PubSub.broadcast(
WhoNeedHelp.PubSub,

189
lib/who_need_help/tracking.ex
View File

@ -9,7 +9,7 @@ defmodule WhoNeedHelp.Tracking do
"""
import Ecto.Query
alias Ecto.Multi
alias WhoNeedHelp.Accounts
alias WhoNeedHelp.Accounts.Scope
alias WhoNeedHelp.Help
alias WhoNeedHelp.Help.Assignment
@ -21,15 +21,11 @@ defmodule WhoNeedHelp.Tracking do
Phoenix.PubSub.subscribe(WhoNeedHelp.PubSub, "tracking:#{assignment_id}")
end
def active_session?(%Scope{user: user} = scope, %Assignment{} = assignment) do
Help.participant?(scope, assignment) &&
Repo.exists?(
from session in TrackingSession,
where:
session.assignment_id == ^assignment.id and session.user_id == ^user.id and
session.active
)
end
def active_session?(%Scope{user: user} = scope, %Assignment{} = assignment),
do: not is_nil(active_session(scope, assignment, user.id))
def active_session(%Scope{user: user} = scope, %Assignment{} = assignment),
do: active_session(scope, assignment, user.id)
def start_session(
%Scope{user: user} = scope,
@ -97,43 +93,31 @@ defmodule WhoNeedHelp.Tracking do
def stop_session(%Scope{user: user} = scope, %Assignment{} = assignment) do
if Help.participant?(scope, assignment) do
case Repo.get_by(TrackingSession,
assignment_id: assignment.id,
user_id: user.id,
active: true
) do
nil ->
{:ok, :already_stopped}
session ->
now = DateTime.utc_now(:second)
Multi.new()
|> Multi.delete_all(:positions, where(Position, tracking_session_id: ^session.id))
|> Multi.update(
:session,
TrackingSession.changeset(session, %{active: false, ended_at: now})
)
|> Repo.transaction()
|> case do
{:ok, _} ->
Phoenix.PubSub.broadcast(
WhoNeedHelp.PubSub,
"tracking:#{assignment.id}",
{:tracking_stopped, user.id}
)
{:ok, :stopped}
{:error, _step, reason, _} ->
{:error, reason}
end
end
stop_locked_session(assignment.id, user.id)
else
{:error, :forbidden}
end
end
def stop_browser_session(assignment_id, user_id, tracking_session_id) do
with {:ok, assignment_id} <- Ecto.UUID.cast(assignment_id),
{:ok, user_id} <- Ecto.UUID.cast(user_id),
{:ok, tracking_session_id} <- Ecto.UUID.cast(tracking_session_id),
%Assignment{} = assignment <-
Assignment
|> Repo.get(assignment_id)
|> Repo.preload(:request),
%Accounts.User{} = user <- Repo.get(Accounts.User, user_id) do
if Help.participant?(Scope.for_user(user), assignment) do
stop_locked_session(assignment.id, user.id, tracking_session_id)
else
{:error, :forbidden}
end
else
_missing_or_invalid -> {:ok, :already_stopped}
end
end
def list_current_positions(%Scope{} = scope, %Assignment{} = assignment) do
if Trust.eligible?(scope) and Help.participant?(scope, assignment) do
TrackingSession
@ -176,6 +160,67 @@ defmodule WhoNeedHelp.Tracking do
end)
end
defp active_session(scope, assignment, user_id) do
if Help.participant?(scope, assignment) do
Repo.get_by(TrackingSession,
assignment_id: assignment.id,
user_id: user_id,
active: true
)
end
end
defp stop_locked_session(assignment_id, user_id, tracking_session_id \\ nil) do
result =
Repo.transact(fn ->
query =
from session in TrackingSession,
where:
session.assignment_id == ^assignment_id and session.user_id == ^user_id and
session.active,
lock: "FOR UPDATE"
query =
if tracking_session_id,
do: where(query, [session], session.id == ^tracking_session_id),
else: query
case Repo.one(query) do
nil ->
{:ok, :already_stopped}
session ->
Position
|> where([position], position.tracking_session_id == ^session.id)
|> Repo.delete_all()
now = DateTime.utc_now(:second)
case Repo.update(TrackingSession.changeset(session, %{active: false, ended_at: now})) do
{:ok, _session} -> {:ok, :stopped}
{:error, changeset} -> {:error, changeset}
end
end
end)
case result do
{:ok, :stopped} ->
Phoenix.PubSub.broadcast(
WhoNeedHelp.PubSub,
"tracking:#{assignment_id}",
{:tracking_stopped, user_id}
)
{:ok, :stopped}
{:ok, :already_stopped} ->
{:ok, :already_stopped}
{:error, reason} ->
{:error, reason}
end
end
defp public_position(%Position{
position: %Geo.Point{coordinates: {lng, lat}},
accuracy_meters: accuracy,
@ -188,7 +233,6 @@ defmodule WhoNeedHelp.Tracking do
previous =
Position
|> where([position], position.tracking_session_id == ^session.id)
|> lock("FOR UPDATE")
|> Repo.one()
now = DateTime.utc_now(:second)
@ -214,21 +258,9 @@ defmodule WhoNeedHelp.Tracking do
})
|> Repo.update()
assignment =
Assignment
|> where([candidate], candidate.id == ^assignment.id)
|> lock("FOR UPDATE")
|> Repo.one!()
assignment =
maybe_mark_helper_movement(assignment, user.id, session.movement_observed_at, now)
assignment = maybe_mark_proximity(assignment, session, position, now)
evidence = %{
helper_movement_observed_at: assignment.helper_movement_observed_at,
proximity_observed_at: assignment.proximity_observed_at
}
maybe_mark_proximity(assignment, session, position, now)
evidence = assignment_evidence(assignment.id)
{:ok, {position, evidence}}
end
@ -237,17 +269,21 @@ defmodule WhoNeedHelp.Tracking do
defp maybe_mark_helper_movement(assignment, user_id, movement_observed_at, now) do
if assignment.helper_id == user_id and not is_nil(movement_observed_at) and
is_nil(assignment.helper_movement_observed_at) do
assignment
|> Assignment.changeset(%{helper_movement_observed_at: now})
|> Repo.update!()
else
assignment
end
Assignment
|> where(
[candidate],
candidate.id == ^assignment.id and
is_nil(candidate.helper_movement_observed_at)
)
|> Repo.update_all(set: [helper_movement_observed_at: now, updated_at: now])
end
defp maybe_mark_proximity(%Assignment{proximity_observed_at: observed} = assignment, _, _, _)
:ok
end
defp maybe_mark_proximity(%Assignment{proximity_observed_at: observed}, _, _, _)
when not is_nil(observed),
do: assignment
do: :ok
defp maybe_mark_proximity(assignment, session, position, now) do
counterpart_positions =
@ -262,12 +298,25 @@ defmodule WhoNeedHelp.Tracking do
|> Repo.all()
if Enum.any?(counterpart_positions, &accuracy_envelopes_overlap?(&1, position)) do
assignment
|> Assignment.changeset(%{proximity_observed_at: now})
|> Repo.update!()
else
assignment
Assignment
|> where(
[candidate],
candidate.id == ^assignment.id and is_nil(candidate.proximity_observed_at)
)
|> Repo.update_all(set: [proximity_observed_at: now, updated_at: now])
end
:ok
end
defp assignment_evidence(assignment_id) do
Assignment
|> where([assignment], assignment.id == ^assignment_id)
|> select([assignment], %{
helper_movement_observed_at: assignment.helper_movement_observed_at,
proximity_observed_at: assignment.proximity_observed_at
})
|> Repo.one!()
end
defp movement_evidence(nil, _current), do: 0.0

32
lib/who_need_help/tracking_presence_cleanup.ex Normal file
View File

@ -0,0 +1,32 @@
defmodule WhoNeedHelp.TrackingPresenceCleanup do
@moduledoc false
use GenServer
alias WhoNeedHelp.Tracking
alias WhoNeedHelpWeb.Presence
def start_link(options), do: GenServer.start_link(__MODULE__, options, name: __MODULE__)
def maybe_stop(key, assignment_id, user_id, tracking_session_id) do
GenServer.cast(
__MODULE__,
{:maybe_stop, key, assignment_id, user_id, tracking_session_id}
)
end
@impl GenServer
def init(_options), do: {:ok, %{}}
@impl GenServer
def handle_cast(
{:maybe_stop, key, assignment_id, user_id, tracking_session_id},
state
) do
if Presence.get_by_key(Presence.tracking_topic(), key) == [] do
Tracking.stop_browser_session(assignment_id, user_id, tracking_session_id)
end
{:noreply, state}
end
end

93
lib/who_need_help/trust.ex
View File

@ -25,13 +25,7 @@ defmodule WhoNeedHelp.Trust do
}
def authorize_action(%Scope{user: %User{} = user}, action) do
user = Repo.get(User, user.id)
cond do
not Accounts.eligible_for_trust_actions?(user) ->
{:error, :account_not_eligible}
true ->
if Accounts.eligible_user_id?(user.id) do
case RateLimiter.check(action, user.id) do
{:error, :rate_limited} = error ->
create_velocity_signal_once(user.id, action)
@ -40,6 +34,8 @@ defmodule WhoNeedHelp.Trust do
result ->
result
end
else
{:error, :account_not_eligible}
end
end
@ -107,26 +103,45 @@ defmodule WhoNeedHelp.Trust do
def paginate_visible_reviews(user_id, options \\ []) do
limit = Pagination.limit(options)
cursor = Pagination.cursor(options)
public_user = Accounts.public_user_query()
Review
|> where([review], review.reviewee_id == ^user_id and not is_nil(review.revealed_at))
|> before_review(cursor)
|> order_by([review], desc: review.inserted_at, desc: review.id)
|> limit(^(limit + 1))
|> preload(:reviewer)
|> preload([review], reviewer: ^public_user)
|> Repo.all()
|> Pagination.page(limit, &{&1.inserted_at, &1.id})
end
def reputation(user_id) do
@empty_reputation %{
completed: 0,
unique_people: 0,
verified_handovers: 0,
location_supported: 0,
rating: nil
}
def reputation(user_id), do: Map.fetch!(reputations([user_id]), user_id)
def reputations([]), do: %{}
def reputations(user_ids) when is_list(user_ids) do
user_ids = Enum.uniq(user_ids)
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)
|> where(
[assignment],
assignment.status == :completed and assignment.helper_id in ^user_ids
)
|> select([assignment, request], %{
id: assignment.id,
user_id: assignment.helper_id,
counterpart_id: request.requester_id,
verified: not is_nil(assignment.handover_verified_at),
location_supported:
@ -141,29 +156,64 @@ defmodule WhoNeedHelp.Trust do
)
|> where(
[assignment, request],
assignment.status == :completed and request.requester_id == ^user_id and
assignment.helper_id != ^user_id
assignment.status == :completed and request.requester_id in ^user_ids and
assignment.helper_id != request.requester_id
)
|> select([assignment], %{
|> select([assignment, request], %{
id: assignment.id,
user_id: request.requester_id,
counterpart_id: assignment.helper_id,
verified: not is_nil(assignment.handover_verified_at),
location_supported: not is_nil(assignment.proximity_observed_at)
})
aggregate =
aggregates =
helper_rows
|> union_all(^requester_rows)
|> subquery()
|> group_by([row], row.user_id)
|> select([row], %{
user_id: row.user_id,
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))
ratings =
Review
|> where(
[review],
review.reviewee_id in ^user_ids and not is_nil(review.revealed_at)
)
|> group_by([review], review.reviewee_id)
|> select([review], %{user_id: review.reviewee_id, rating: avg(review.rating)})
rows =
aggregates
|> subquery()
|> join(:left, [aggregate], rating in subquery(ratings),
on: rating.user_id == aggregate.user_id
)
|> select([aggregate, rating], %{
user_id: aggregate.user_id,
completed: aggregate.completed,
unique_people: aggregate.unique_people,
verified_handovers: aggregate.verified_handovers,
location_supported: aggregate.location_supported,
rating: rating.rating
})
|> Repo.all()
Enum.reduce(rows, Map.new(user_ids, &{&1, @empty_reputation}), fn row, reputations ->
Map.put(
reputations,
row.user_id,
row
|> Map.delete(:user_id)
|> Map.update!(:rating, &decimal_average/1)
)
end)
end
def leaderboard do
@ -389,13 +439,14 @@ defmodule WhoNeedHelp.Trust do
def paginate_blocks(%Scope{user: user}, options \\ []) do
limit = Pagination.limit(options)
cursor = Pagination.cursor(options)
public_user = Accounts.public_user_query()
Block
|> where([block], block.blocker_id == ^user.id)
|> before_block(cursor)
|> order_by([block], desc: block.inserted_at, desc: block.id)
|> limit(^(limit + 1))
|> preload(:blocked)
|> preload([block], blocked: ^public_user)
|> Repo.all()
|> Pagination.page(limit, &{&1.inserted_at, &1.id})
end
@ -685,14 +736,6 @@ defmodule WhoNeedHelp.Trust do
|> Repo.insert()
end
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

78
lib/who_need_help/workers/expire_requests.ex
View File

@ -12,44 +12,8 @@ defmodule WhoNeedHelp.Workers.ExpireRequests do
def perform(_job) do
now = DateTime.utc_now(:second)
ids =
HelpRequest
|> where([r], r.status == :open and r.expires_at <= ^now)
|> select([request], request.id)
|> Repo.all()
expired =
Enum.count(ids, fn id ->
Repo.transact(fn ->
request =
HelpRequest
|> where(
[request],
request.id == ^id and request.status == :open and request.expires_at <= ^now
)
|> lock("FOR UPDATE")
|> Repo.one()
if request do
with {:ok, request} <-
request
|> Ecto.Changeset.change(status: :expired)
|> Repo.update(),
{:ok, _audit} <-
Trust.audit(nil, "request.expired", "request", request.id) do
{:ok, request}
end
else
{:error, :already_transitioned}
end
end)
|> case do
{:ok, _request} -> true
_other -> false
end
end)
{:ok, cleanup} = Tracking.cleanup_finished_sessions()
with {:ok, expired} <- expire_available(now, 0),
{:ok, cleanup} <- Tracking.cleanup_finished_sessions() do
{pruned_buckets, _} = RateLimiter.prune_expired()
{:ok,
@ -60,4 +24,42 @@ defmodule WhoNeedHelp.Workers.ExpireRequests do
rate_limit_buckets_pruned: pruned_buckets
}}
end
end
# Claim a single row per short transaction. This keeps memory bounded and
# allows another worker to make progress without waiting for a large backlog
# lock. The recursion is tail-recursive and stops on the first empty claim.
defp expire_available(now, expired) do
result =
Repo.transact(fn ->
request =
HelpRequest
|> where([request], request.status == :open and request.expires_at <= ^now)
|> order_by([request], asc: request.expires_at, asc: request.id)
|> limit(1)
|> lock("FOR UPDATE SKIP LOCKED")
|> Repo.one()
case request do
nil ->
{:ok, :empty}
request ->
with {:ok, request} <-
request
|> Ecto.Changeset.change(status: :expired)
|> Repo.update(),
{:ok, _audit} <-
Trust.audit(nil, "request.expired", "request", request.id) do
{:ok, request}
end
end
end)
case result do
{:ok, :empty} -> {:ok, expired}
{:ok, %HelpRequest{}} -> expire_available(now, expired + 1)
{:error, reason} -> {:error, reason}
end
end
end

12
lib/who_need_help_web/controllers/metrics_controller.ex
View File

@ -1,6 +1,8 @@
defmodule WhoNeedHelpWeb.MetricsController do
use WhoNeedHelpWeb, :controller
alias WhoNeedHelpWeb.MetricsAccess
@content_type "text/plain; version=0.0.4"
# Sobelow's HTML-oriented SendResp check cannot infer this fixed Prometheus
@ -9,7 +11,7 @@ defmodule WhoNeedHelpWeb.MetricsController do
def show(conn, _params) do
token = Application.get_env(:who_need_help, :metrics_token)
if authorized?(get_req_header(conn, "authorization"), token) do
if MetricsAccess.authorized?(get_req_header(conn, "authorization"), token) do
body = TelemetryMetricsPrometheus.Core.scrape(:prometheus_metrics)
conn
@ -23,12 +25,4 @@ defmodule WhoNeedHelpWeb.MetricsController do
|> send_resp(:unauthorized, "Unauthorized\n")
end
end
defp authorized?(["Bearer " <> candidate], expected)
when is_binary(expected) and expected != "" and
byte_size(candidate) == byte_size(expected) do
Plug.Crypto.secure_compare(candidate, expected)
end
defp authorized?(_authorization, _expected), do: false
end

2
lib/who_need_help_web/live/activity_live/index.ex
View File

@ -175,7 +175,7 @@ defmodule WhoNeedHelpWeb.ActivityLive.Index do
)}
</span>
<span class="badge badge-outline">
{Enum.count(activity.participants, &(&1.status == :approved))}/{activity.capacity}
{activity.approved_participant_count}/{activity.capacity}
</span>
</div>
<h2 class="text-xl font-bold">{activity.title}</h2>

2
lib/who_need_help_web/live/category_proposal_live.ex
View File

@ -148,7 +148,7 @@ defmodule WhoNeedHelpWeb.CategoryProposalLive do
</p>
</div>
<button phx-click="vote" phx-value-id={proposal.id} class="btn btn-outline btn-sm">
▲ {length(proposal.votes)}
▲ {proposal.vote_count}
</button>
</div>
</article>

2
lib/who_need_help_web/live/moderation_live.ex
View File

@ -478,7 +478,7 @@ defmodule WhoNeedHelpWeb.ModerationLive do
</h3>
<p class="mt-1 text-sm">{proposal.reason}</p>
<p class="mt-1 text-xs text-base-content/50">
{gettext("%{count} community votes", count: length(proposal.votes))}
{gettext("%{count} community votes", count: proposal.vote_count)}
</p>
<div :if={proposal.status == :open} class="mt-4 grid gap-4 xl:grid-cols-2">
<.form

92
lib/who_need_help_web/live/request_live/show.ex
View File

@ -2,6 +2,7 @@ defmodule WhoNeedHelpWeb.RequestLive.Show do
use WhoNeedHelpWeb, :live_view
alias WhoNeedHelp.{Help, Messaging, Tracking, Trust}
alias WhoNeedHelpWeb.Presence
@e2e_routes Application.compile_env(:who_need_help, :e2e_routes, false)
@ -16,6 +17,8 @@ defmodule WhoNeedHelpWeb.RequestLive.Show do
socket =
socket
|> assign(:subscribed_assignment_id, nil)
|> assign(:tracking_presence_key, nil)
|> assign(:tracking_session_id, nil)
|> assign(:native_client, native_client)
|> assign(
:e2e_runtime_node,
@ -33,13 +36,18 @@ defmodule WhoNeedHelpWeb.RequestLive.Show do
socket = maybe_subscribe_assignment(socket, request)
tracking_active =
tracking_session =
request.assignment &&
Tracking.active_session?(socket.assigns.current_scope, request.assignment)
Tracking.active_session(socket.assigns.current_scope, request.assignment)
socket =
socket
|> load(request, tracking_active)
|> assign(
:tracking_session_id,
if(tracking_session, do: tracking_session.id, else: nil)
)
|> load(request, not is_nil(tracking_session))
|> maybe_track_browser_presence()
|> maybe_start_native_tracking()
{:ok, socket}
@ -88,7 +96,10 @@ defmodule WhoNeedHelpWeb.RequestLive.Show do
|> assign_positions(positions)
|> then(fn socket ->
if user_id == socket.assigns.current_scope.user.id do
assign(socket, :tracking_active, false)
socket
|> assign(:tracking_active, false)
|> assign(:tracking_session_id, nil)
|> maybe_untrack_browser_presence()
else
socket
end
@ -171,10 +182,12 @@ defmodule WhoNeedHelpWeb.RequestLive.Show do
def handle_event("start-tracking", _, socket) do
case Tracking.start_session(socket.assigns.current_scope, socket.assigns.assignment) do
{:ok, _} ->
{:ok, tracking_session} ->
{:noreply,
socket
|> assign(:tracking_active, true)
|> assign(:tracking_session_id, tracking_session.id)
|> maybe_track_browser_presence()
|> maybe_start_native_tracking()}
{:error, reason} ->
@ -199,6 +212,8 @@ defmodule WhoNeedHelpWeb.RequestLive.Show do
{:noreply,
socket
|> assign(:tracking_active, false)
|> assign(:tracking_session_id, nil)
|> maybe_untrack_browser_presence()
|> maybe_stop_native_tracking()}
{:error, reason} ->
@ -268,13 +283,19 @@ defmodule WhoNeedHelpWeb.RequestLive.Show do
def handle_event("block-user", _, socket) do
case Trust.block(socket.assigns.current_scope, socket.assigns.other_user_id) do
{:ok, _block} ->
if socket.assigns.tracking_active and socket.assigns.assignment do
Tracking.stop_session(socket.assigns.current_scope, socket.assigns.assignment)
end
{:noreply,
socket
|> maybe_untrack_browser_presence()
|> assign(:blocked_by_current, true)
|> assign(:messages, [])
|> assign(:messages_cursor, nil)
|> assign(:positions, %{})
|> assign(:tracking_active, false)
|> assign(:tracking_session_id, nil)
|> assign(
:markers,
Jason.encode!(request_markers(socket.assigns.current_scope, socket.assigns.request))
@ -302,25 +323,6 @@ defmodule WhoNeedHelpWeb.RequestLive.Show do
end
end
@impl true
def terminate(_reason, socket) do
case socket.assigns do
%{
tracking_active: true,
native_client: false,
current_scope: scope,
assignment: assignment
}
when not is_nil(assignment) ->
Tracking.stop_session(scope, assignment)
_ ->
:ok
end
:ok
end
defp maybe_start_native_tracking(
%{assigns: %{native_client: true, tracking_active: true, assignment: assignment}} =
socket
@ -337,6 +339,39 @@ defmodule WhoNeedHelpWeb.RequestLive.Show do
defp maybe_stop_native_tracking(socket), do: socket
defp maybe_track_browser_presence(
%{
assigns: %{
native_client: false,
tracking_active: true,
tracking_presence_key: nil,
tracking_session_id: tracking_session_id,
assignment: assignment,
current_scope: %{user: user}
}
} = socket
)
when not is_nil(assignment) do
if connected?(socket) do
case Presence.track_browser(self(), assignment.id, user.id, tracking_session_id) do
{:ok, key} -> assign(socket, :tracking_presence_key, key)
_error -> socket
end
else
socket
end
end
defp maybe_track_browser_presence(socket), do: socket
defp maybe_untrack_browser_presence(%{assigns: %{tracking_presence_key: key}} = socket)
when is_binary(key) do
Presence.untrack_browser(self(), key)
assign(socket, :tracking_presence_key, nil)
end
defp maybe_untrack_browser_presence(socket), do: socket
defp transition(socket, fun, success) do
case fun.(socket.assigns.current_scope, socket.assigns.assignment.id) do
{:ok, _} -> {:noreply, put_flash(socket, :info, success)}
@ -366,6 +401,11 @@ defmodule WhoNeedHelpWeb.RequestLive.Show do
do: Tracking.list_current_positions(socket.assigns.current_scope, assignment),
else: %{}
reputations =
[request.requester_id, assignment && assignment.helper_id]
|> Enum.reject(&is_nil/1)
|> Trust.reputations()
socket
|> assign(:page_title, request.title)
|> assign(:request, request)
@ -373,10 +413,10 @@ defmodule WhoNeedHelpWeb.RequestLive.Show do
:structured_details,
structured_details(request, Gettext.get_locale(WhoNeedHelpWeb.Gettext))
)
|> assign(:requester_reputation, Trust.reputation(request.requester_id))
|> assign(:requester_reputation, Map.fetch!(reputations, request.requester_id))
|> assign(
:helper_reputation,
if(assignment, do: Trust.reputation(assignment.helper_id), else: nil)
if(assignment, do: Map.fetch!(reputations, assignment.helper_id), else: nil)
)
|> assign(:assignment, assignment)
|> assign(:participant, participant)

11
lib/who_need_help_web/metrics_access.ex Normal file
View File

@ -0,0 +1,11 @@
defmodule WhoNeedHelpWeb.MetricsAccess do
@moduledoc false
def authorized?(["Bearer " <> candidate], expected)
when is_binary(expected) and expected != "" and
byte_size(candidate) == byte_size(expected) do
Plug.Crypto.secure_compare(candidate, expected)
end
def authorized?(_authorization, _expected), do: false
end

53
lib/who_need_help_web/presence.ex
View File

@ -2,4 +2,57 @@ defmodule WhoNeedHelpWeb.Presence do
use Phoenix.Presence,
otp_app: :who_need_help,
pubsub_server: WhoNeedHelp.PubSub
@tracking_topic "browser-tracking-presence"
def tracking_topic, do: @tracking_topic
def tracking_key(assignment_id, user_id), do: "#{assignment_id}:#{user_id}"
def track_browser(pid, assignment_id, user_id, tracking_session_id) do
key = tracking_key(assignment_id, user_id)
case track(pid, @tracking_topic, key, %{
assignment_id: assignment_id,
user_id: user_id,
tracking_session_id: tracking_session_id
}) do
{:ok, _ref} -> {:ok, key}
{:error, {:already_tracked, _pid, @tracking_topic, ^key}} -> {:ok, key}
other -> other
end
end
def untrack_browser(pid, key), do: untrack(pid, @tracking_topic, key)
@impl Phoenix.Presence
def init(_options), do: {:ok, %{}}
@impl Phoenix.Presence
def handle_metas(@tracking_topic, %{leaves: leaves}, presences, state) do
Enum.each(leaves, fn {key, %{metas: metas}} ->
if not Map.has_key?(presences, key) do
case List.first(metas) do
%{
assignment_id: assignment_id,
user_id: user_id,
tracking_session_id: tracking_session_id
} ->
WhoNeedHelp.TrackingPresenceCleanup.maybe_stop(
key,
assignment_id,
user_id,
tracking_session_id
)
_other ->
:ok
end
end
end)
{:ok, state}
end
def handle_metas(_topic, _diff, _presences, state), do: {:ok, state}
end

81
lib/who_need_help_web/telemetry.ex
View File

@ -9,7 +9,7 @@ defmodule WhoNeedHelpWeb.Telemetry do
@impl true
def init(_arg) do
reporter_children =
if Application.fetch_env!(:who_need_help, :app_role) == :web do
if Application.fetch_env!(:who_need_help, :app_role) in [:web, :worker] do
[
{TelemetryMetricsPrometheus.Core,
name: :prometheus_metrics, metrics: prometheus_metrics(), start_async: false}
@ -43,6 +43,15 @@ defmodule WhoNeedHelpWeb.Telemetry do
end,
description: "Cumulative HTTP request duration"
),
distribution("who_need_help.http.request.duration.seconds",
event_name: [:phoenix, :endpoint, :stop],
measurement: :duration,
unit: {:native, :second},
reporter_options: [
buckets: [0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0]
],
description: "HTTP request duration distribution"
),
counter("who_need_help.http.exceptions.total",
event_name: [:phoenix, :router_dispatch, :exception],
measurement: :duration,
@ -76,6 +85,48 @@ defmodule WhoNeedHelpWeb.Telemetry do
end,
description: "Cumulative time waiting for a database connection"
),
distribution("who_need_help.database.query.duration.seconds",
event_name: [:who_need_help, :repo, :query],
measurement: :total_time,
unit: {:native, :second},
reporter_options: [
buckets: [
0.000_05,
0.000_1,
0.000_25,
0.000_5,
0.001,
0.0025,
0.005,
0.01,
0.025,
0.05,
0.1
]
],
description: "Database query duration distribution"
),
distribution("who_need_help.database.query.queue.duration.seconds",
event_name: [:who_need_help, :repo, :query],
measurement: fn measurements -> Map.get(measurements, :queue_time, 0) end,
unit: {:native, :second},
reporter_options: [
buckets: [
0.000_01,
0.000_025,
0.000_05,
0.000_1,
0.000_25,
0.000_5,
0.001,
0.0025,
0.005,
0.01,
0.025
]
],
description: "Time waiting for an Ecto connection"
),
sum("who_need_help.database.query.decode.duration.microseconds.total",
event_name: [:who_need_help, :repo, :query],
measurement: fn measurements ->
@ -90,6 +141,34 @@ defmodule WhoNeedHelpWeb.Telemetry do
measurement: :duration,
description: "Completed WebSocket connection attempts"
),
counter("who_need_help.oban.jobs.completed.total",
event_name: [:oban, :job, :stop],
measurement: :duration,
tags: [:queue],
description: "Completed Oban jobs"
),
counter("who_need_help.oban.jobs.failed.total",
event_name: [:oban, :job, :exception],
measurement: :duration,
tags: [:queue],
description: "Failed Oban job attempts"
),
sum("who_need_help.oban.job.duration.microseconds.total",
event_name: [:oban, :job, :stop],
measurement: fn measurements ->
System.convert_time_unit(measurements.duration, :native, :microsecond)
end,
tags: [:queue],
description: "Cumulative successful Oban execution duration"
),
sum("who_need_help.oban.job.queue.duration.microseconds.total",
event_name: [:oban, :job, :stop],
measurement: fn measurements ->
System.convert_time_unit(measurements.queue_time, :native, :microsecond)
end,
tags: [:queue],
description: "Cumulative Oban queue wait duration"
),
last_value("who_need_help.vm.memory.total.bytes",
event_name: [:vm, :memory],
measurement: :total,

76
lib/who_need_help_web/worker_metrics_plug.ex Normal file
View File

@ -0,0 +1,76 @@
defmodule WhoNeedHelpWeb.WorkerMetricsPlug do
@moduledoc """
Internal-only health and Prometheus endpoint for worker-role releases.
Compose and Kubernetes keep this listener off public ingress. Metrics still
require the same independent bearer credential as web replicas.
"""
import Plug.Conn
alias WhoNeedHelpWeb.MetricsAccess
@content_type "text/plain; version=0.0.4"
def init(options), do: options
def call(%Plug.Conn{method: "GET", path_info: ["healthz", "live"]} = conn, _options) do
json(conn, :ok, %{status: "ok", node: to_string(Node.self()), role: "worker"})
end
def call(%Plug.Conn{method: "GET", path_info: ["healthz", "ready"]} = conn, _options) do
database_ready? =
match?({:ok, _}, Ecto.Adapters.SQL.query(WhoNeedHelp.Repo, "SELECT 1", []))
if database_ready? and is_pid(Oban.whereis(Oban)) do
json(conn, :ok, %{status: "ready", node: to_string(Node.self()), role: "worker"})
else
json(conn, :service_unavailable, %{status: "not_ready", role: "worker"})
end
end
# Sobelow's HTML-oriented SendResp check cannot infer this fixed Prometheus
# text content type. The response is generated by the metrics registry and
# is not rendered in an HTML context.
# sobelow_skip ["XSS.SendResp"]
def call(%Plug.Conn{method: "GET", path_info: ["metrics"]} = conn, _options) do
token = Application.get_env(:who_need_help, :metrics_token)
if MetricsAccess.authorized?(get_req_header(conn, "authorization"), token) do
body = TelemetryMetricsPrometheus.Core.scrape(:prometheus_metrics)
conn
|> put_resp_header("cache-control", "no-store")
|> put_resp_content_type(@content_type)
|> send_resp(:ok, body)
else
conn
|> put_resp_header("cache-control", "no-store")
|> put_resp_header("www-authenticate", "Bearer")
|> send_resp(:unauthorized, "Unauthorized\n")
end
end
def call(conn, _options), do: send_resp(conn, :not_found, "Not found\n")
def child_spec(_options) do
endpoint_config = Application.fetch_env!(:who_need_help, WhoNeedHelpWeb.Endpoint)
port = endpoint_config |> Keyword.fetch!(:http) |> Keyword.fetch!(:port)
Bandit.child_spec(
plug: __MODULE__,
scheme: :http,
ip: {0, 0, 0, 0},
port: port,
startup_log: false
)
|> Supervisor.child_spec(id: __MODULE__)
end
defp json(conn, status, body) do
conn
|> put_resp_header("cache-control", "no-store")
|> put_resp_content_type("application/json")
|> send_resp(status, Jason.encode!(body))
end
end

196
ops/observability/grafana/dashboards/who-need-help-overview.json
View File

@ -173,6 +173,202 @@
],
"title": "Database query rate by replica",
"type": "timeseries"
},
{
"datasource": {
"type": "prometheus",
"uid": "wnh-prometheus"
},
"gridPos": {
"h": 8,
"w": 8,
"x": 0,
"y": 16
},
"id": 5,
"targets": [
{
"editorMode": "code",
"expr": "sum(up{job=\"who-need-help-worker\"})",
"legendFormat": "available",
"range": true,
"refId": "A"
}
],
"title": "Available worker replicas",
"type": "stat"
},
{
"datasource": {
"type": "prometheus",
"uid": "wnh-prometheus"
},
"fieldConfig": {
"defaults": {
"unit": "s"
},
"overrides": []
},
"gridPos": {
"h": 8,
"w": 16,
"x": 8,
"y": 16
},
"id": 6,
"targets": [
{
"editorMode": "code",
"expr": "histogram_quantile(0.95, sum by (le, instance) (rate(who_need_help_http_request_duration_seconds_bucket{job=\"who-need-help-web\"}[1m])))",
"legendFormat": "p95 {{instance}}",
"range": true,
"refId": "A"
},
{
"editorMode": "code",
"expr": "histogram_quantile(0.99, sum by (le, instance) (rate(who_need_help_http_request_duration_seconds_bucket{job=\"who-need-help-web\"}[1m])))",
"legendFormat": "p99 {{instance}}",
"range": true,
"refId": "B"
}
],
"title": "HTTP request latency by web replica",
"type": "timeseries"
},
{
"datasource": {
"type": "prometheus",
"uid": "wnh-prometheus"
},
"fieldConfig": {
"defaults": {
"unit": "s"
},
"overrides": []
},
"gridPos": {
"h": 8,
"w": 12,
"x": 0,
"y": 24
},
"id": 7,
"targets": [
{
"editorMode": "code",
"expr": "histogram_quantile(0.95, sum by (le, instance) (rate(who_need_help_database_query_duration_seconds_bucket[1m])))",
"legendFormat": "query p95 {{instance}}",
"range": true,
"refId": "A"
},
{
"editorMode": "code",
"expr": "histogram_quantile(0.95, sum by (le, instance) (rate(who_need_help_database_query_queue_duration_seconds_bucket[1m])))",
"legendFormat": "pool wait p95 {{instance}}",
"range": true,
"refId": "B"
}
],
"title": "Database query and pool-wait latency",
"type": "timeseries"
},
{
"datasource": {
"type": "prometheus",
"uid": "wnh-prometheus"
},
"fieldConfig": {
"defaults": {
"unit": "ops"
},
"overrides": []
},
"gridPos": {
"h": 8,
"w": 12,
"x": 12,
"y": 24
},
"id": 8,
"targets": [
{
"editorMode": "code",
"expr": "sum by (queue) (rate(who_need_help_oban_jobs_completed_total{job=\"who-need-help-worker\"}[1m]))",
"legendFormat": "completed {{queue}}",
"range": true,
"refId": "A"
},
{
"editorMode": "code",
"expr": "sum by (queue) (rate(who_need_help_oban_jobs_failed_total{job=\"who-need-help-worker\"}[1m]))",
"legendFormat": "failed {{queue}}",
"range": true,
"refId": "B"
}
],
"title": "Oban job outcomes by queue",
"type": "timeseries"
},
{
"datasource": {
"type": "prometheus",
"uid": "wnh-prometheus"
},
"fieldConfig": {
"defaults": {
"unit": "ms"
},
"overrides": []
},
"gridPos": {
"h": 8,
"w": 12,
"x": 0,
"y": 32
},
"id": 9,
"targets": [
{
"editorMode": "code",
"expr": "sum by (queue) (rate(who_need_help_oban_job_duration_microseconds_total{job=\"who-need-help-worker\"}[5m])) / sum by (queue) (rate(who_need_help_oban_jobs_completed_total{job=\"who-need-help-worker\"}[5m])) / 1000",
"legendFormat": "execution {{queue}}",
"range": true,
"refId": "A"
},
{
"editorMode": "code",
"expr": "sum by (queue) (rate(who_need_help_oban_job_queue_duration_microseconds_total{job=\"who-need-help-worker\"}[5m])) / sum by (queue) (rate(who_need_help_oban_jobs_completed_total{job=\"who-need-help-worker\"}[5m])) / 1000",
"legendFormat": "queue wait {{queue}}",
"range": true,
"refId": "B"
}
],
"title": "Average Oban execution and queue wait",
"type": "timeseries"
},
{
"datasource": {
"type": "prometheus",
"uid": "wnh-prometheus"
},
"gridPos": {
"h": 8,
"w": 12,
"x": 12,
"y": 32
},
"id": 10,
"targets": [
{
"editorMode": "code",
"expr": "who_need_help_vm_run_queue_total",
"legendFormat": "{{job}} {{instance}}",
"range": true,
"refId": "A"
}
],
"title": "BEAM scheduler run queue by replica",
"type": "timeseries"
}
],
"refresh": "5s",

10
ops/observability/prometheus.template.yml
View File

@ -25,3 +25,13 @@ scrape_configs:
- files:
- /runtime/web-targets.json
refresh_interval: __SCRAPE_INTERVAL__
- job_name: who-need-help-worker
metrics_path: /metrics
authorization:
type: Bearer
credentials_file: /runtime/metrics-token
file_sd_configs:
- files:
- /runtime/worker-targets.json
refresh_interval: __SCRAPE_INTERVAL__

7
ops/observability/rules.yml
View File

@ -8,3 +8,10 @@ groups:
annotations:
summary: "The local Prometheus scraper cannot reach one web replica"
description: "The direct metrics target {{ $labels.instance }} is unavailable."
- alert: WhoNeedHelpWorkerReplicaUnavailable
expr: up{job="who-need-help-worker"} == 0
labels:
severity: local-drill
annotations:
summary: "The local Prometheus scraper cannot reach one worker replica"
description: "The direct worker metrics target {{ $labels.instance }} is unavailable."

10
scripts/clean-deploy-verify.sh
View File

@ -136,15 +136,17 @@ secret_key_base=$(openssl rand -hex 64)
handover_secret=$(openssl rand -hex 64)
release_cookie=$(openssl rand -hex 64)
metrics_token=$(openssl rand -hex 32)
docker_socket_gid=$(stat -c '%g' /var/run/docker.sock)
cat >"$env_file" <<EOF
HTTP_PORT=0
MAILPIT_PORT=0
DOCKER_SOCKET_GID=$docker_socket_gid
TRAEFIK_TRUSTED_IPS=127.0.0.1/32
TRAEFIK_RETRY_ATTEMPTS=3
TRAEFIK_PROJECT_CONSTRAINT=$project
TRAEFIK_APP_NAME=wnh-portability-$compact_id
TRAEFIK_DOCKER_NETWORK=${project}_internal
TRAEFIK_DOCKER_NETWORK=${project}_ingress
TRAEFIK_ROUTER_RULE='Host(\`portability.local\`)'
PHX_HOST=portability.local
PHX_SCHEME=http
@ -158,7 +160,11 @@ POSTGRES_DB=$postgres_db
POSTGRES_USER=$postgres_user
POSTGRES_PASSWORD=$postgres_password
DATABASE_URL=ecto://$postgres_user:$postgres_password@db/$postgres_db
POOL_SIZE=10
WEB_POOL_SIZE=4
WORKER_POOL_SIZE=2
MIGRATE_POOL_SIZE=2
WEB_REPLICAS=2
WORKER_REPLICAS=2
ERLANG_PORT_LIMIT=65536
SECRET_KEY_BASE=$secret_key_base
HANDOVER_SECRET=$handover_secret

2
scripts/e2e-run.sh
View File

@ -22,7 +22,7 @@ mkdir -p "$output_dir"
export TRAEFIK_PROJECT_CONSTRAINT="$project"
export TRAEFIK_APP_NAME="who-need-help-e2e-$run_id"
export TRAEFIK_DOCKER_NETWORK="${project}_internal"
export TRAEFIK_DOCKER_NETWORK="${project}_ingress"
export E2E_OUTPUT_DIR="$output_dir"
export E2E_UID
E2E_UID=$(id -u)

8
scripts/ensure-local-e2e-env.sh
View File

@ -40,10 +40,12 @@ secret_key_base=$(openssl rand -hex 64)
handover_secret=$(openssl rand -hex 64)
release_cookie=$(openssl rand -hex 64)
metrics_token=$(openssl rand -hex 32)
docker_socket_gid=$(stat -c '%g' /var/run/docker.sock)
cat >"$target" <<EOF
HTTP_PORT=0
MAILPIT_PORT=0
DOCKER_SOCKET_GID=$docker_socket_gid
TRAEFIK_TRUSTED_IPS=127.0.0.1/32
TRAEFIK_RETRY_ATTEMPTS=3
TRAEFIK_PROJECT_CONSTRAINT=generated-per-run
@ -58,7 +60,11 @@ POSTGRES_DB=who_need_help_e2e
POSTGRES_USER=postgres
POSTGRES_PASSWORD=$postgres_password
DATABASE_URL=ecto://postgres:$postgres_password@db/who_need_help_e2e
POOL_SIZE=10
WEB_POOL_SIZE=4
WORKER_POOL_SIZE=2
MIGRATE_POOL_SIZE=2
WEB_REPLICAS=2
WORKER_REPLICAS=2
ERLANG_PORT_LIMIT=65536
SECRET_KEY_BASE=$secret_key_base
HANDOVER_SECRET=$handover_secret

3
scripts/ensure-local-load-env.sh
View File

@ -301,6 +301,7 @@ backup_minio_root_user="wnh$(openssl rand -hex 12)"
backup_minio_root_password=$(openssl rand -hex 32)
backup_restic_password=$(openssl rand -hex 32)
database_url="ecto://wnh_load:${postgres_password}@db/who_need_help_load"
docker_socket_gid=$(stat -c '%g' /var/run/docker.sock)
temporary=$(mktemp "${ENV_FILE}.XXXXXX")
trap 'rm -f "$temporary"' EXIT HUP INT TERM
@ -315,6 +316,7 @@ OBSERVABILITY_GRAFANA_ADMIN_PASSWORD_VALUE=$observability_grafana_admin_password
BACKUP_MINIO_ROOT_USER_VALUE=$backup_minio_root_user \
BACKUP_MINIO_ROOT_PASSWORD_VALUE=$backup_minio_root_password \
BACKUP_RESTIC_PASSWORD_VALUE=$backup_restic_password \
DOCKER_SOCKET_GID_VALUE=$docker_socket_gid \
perl -0pe '
s/GENERATE_POSTGRES_PASSWORD/$ENV{POSTGRES_PASSWORD_VALUE}/g;
s/GENERATE_DATABASE_URL/$ENV{DATABASE_URL_VALUE}/g;
@ -327,6 +329,7 @@ BACKUP_RESTIC_PASSWORD_VALUE=$backup_restic_password \
s/GENERATE_BACKUP_MINIO_ROOT_USER/$ENV{BACKUP_MINIO_ROOT_USER_VALUE}/g;
s/GENERATE_BACKUP_MINIO_ROOT_PASSWORD/$ENV{BACKUP_MINIO_ROOT_PASSWORD_VALUE}/g;
s/GENERATE_BACKUP_RESTIC_PASSWORD/$ENV{BACKUP_RESTIC_PASSWORD_VALUE}/g;
s/REPLACE_WITH_DOCKER_SOCKET_NUMERIC_GID/$ENV{DOCKER_SOCKET_GID_VALUE}/g;
' "$TEMPLATE" >"$temporary"
if grep -Eq '^[A-Z0-9_]+=GENERATE_' "$temporary"; then

24
scripts/load-run.sh
View File

@ -18,6 +18,9 @@ set -a
. "$ENV_FILE"
set +a
WEB_POOL_SIZE=${WEB_POOL_SIZE:-${POOL_SIZE:-}}
WORKER_POOL_SIZE=${WORKER_POOL_SIZE:-${POOL_SIZE:-}}
duration_source=".env.load"
if [[ -n "$duration_override" ]]; then
LOAD_DURATION=$duration_override
@ -28,7 +31,8 @@ for name in LOAD_PROJECT LOAD_HOST LOAD_WEB_REPLICAS LOAD_WORKER_REPLICAS \
LOAD_HTTP_VUS LOAD_WS_VUS \
LOAD_DURATION LOAD_WS_HOLD_MS LOAD_WS_CONNECT_TIMEOUT_MS \
LOAD_HTTP_THINK_SECONDS LOAD_AUTH_VUS LOAD_AUTH_WS_TIMEOUT_MS \
LOAD_AUTH_THINK_SECONDS LOAD_FIXTURE_PASSWORD POSTGRES_DB HTTP_PORT; do
LOAD_AUTH_THINK_SECONDS LOAD_FIXTURE_PASSWORD POSTGRES_DB HTTP_PORT \
WEB_POOL_SIZE WORKER_POOL_SIZE; do
if [[ -z "${!name:-}" ]]; then
echo "$name is missing from .env.load" >&2
exit 1
@ -36,7 +40,7 @@ for name in LOAD_PROJECT LOAD_HOST LOAD_WEB_REPLICAS LOAD_WORKER_REPLICAS \
done
for name in LOAD_WEB_REPLICAS LOAD_WORKER_REPLICAS LOAD_HTTP_VUS LOAD_WS_VUS \
LOAD_AUTH_VUS; do
LOAD_AUTH_VUS WEB_POOL_SIZE WORKER_POOL_SIZE; do
if [[ ! "${!name}" =~ ^[1-9][0-9]*$ ]]; then
echo "$name must be a positive integer." >&2
exit 1
@ -430,17 +434,23 @@ SQL
summarize_database_connections() {
jq -s \
--argjson pool_size "$POOL_SIZE" \
--argjson app_replicas "$((LOAD_WEB_REPLICAS + LOAD_WORKER_REPLICAS))" '
--argjson web_pool_size "$WEB_POOL_SIZE" \
--argjson worker_pool_size "$WORKER_POOL_SIZE" \
--argjson web_replicas "$LOAD_WEB_REPLICAS" \
--argjson worker_replicas "$LOAD_WORKER_REPLICAS" '
. as $samples |
{
schema_version: 1,
measurement: "read-only pg_stat_activity samples during the k6 run",
thresholds_applied: false,
sample_count: length,
configured_app_replicas: $app_replicas,
configured_pool_size_per_replica: $pool_size,
configured_repo_pool_slots: ($app_replicas * $pool_size),
configured_web_replicas: $web_replicas,
configured_worker_replicas: $worker_replicas,
configured_web_pool_size_per_replica: $web_pool_size,
configured_worker_pool_size_per_replica: $worker_pool_size,
configured_repo_pool_slots:
(($web_replicas * $web_pool_size) +
($worker_replicas * $worker_pool_size)),
max_connections: (map(.max_connections) | max),
reserved_connections: (map(.reserved_connections) | max),
superuser_reserved_connections:

110
scripts/observability-run.sh
View File

@ -15,7 +15,7 @@ set -a
. "$ENV_FILE"
set +a
for name in LOAD_PROJECT LOAD_WEB_REPLICAS POSTGRES_DB METRICS_TOKEN \
for name in LOAD_PROJECT LOAD_WEB_REPLICAS LOAD_WORKER_REPLICAS POSTGRES_DB METRICS_TOKEN \
OBSERVABILITY_PROMETHEUS_PORT OBSERVABILITY_ALERTMANAGER_PORT \
OBSERVABILITY_GRAFANA_PORT OBSERVABILITY_SCRAPE_INTERVAL \
OBSERVABILITY_EVALUATION_INTERVAL OBSERVABILITY_TIMEOUT_SECONDS \
@ -36,10 +36,12 @@ if [[ ! "$LABEL" =~ ^[A-Za-z0-9._-]+$ ]]; then
exit 1
fi
if [[ ! "$LOAD_WEB_REPLICAS" =~ ^[1-9][0-9]*$ ]]; then
echo "LOAD_WEB_REPLICAS must be a positive integer." >&2
for name in LOAD_WEB_REPLICAS LOAD_WORKER_REPLICAS; do
if [[ ! "${!name}" =~ ^[1-9][0-9]*$ ]]; then
echo "$name must be a positive integer." >&2
exit 1
fi
fi
done
for name in OBSERVABILITY_PROMETHEUS_PORT OBSERVABILITY_ALERTMANAGER_PORT \
OBSERVABILITY_GRAFANA_PORT; do
@ -216,23 +218,36 @@ fetch_receiver_events() {
}
wait_for_targets_up() {
local expected=$1
local expected_web=$1
local expected_worker=$2
local deadline=$((SECONDS + OBSERVABILITY_TIMEOUT_SECONDS))
while ((SECONDS < deadline)); do
if curl --fail --silent --show-error \
"$prometheus_url/api/v1/targets?state=active" \
>"$output_dir/targets-current.json" 2>/dev/null &&
jq -e --argjson expected "$expected" --arg run_id "$LABEL" '
jq -e \
--argjson expected_web "$expected_web" \
--argjson expected_worker "$expected_worker" \
--arg run_id "$LABEL" '
[
.data.activeTargets[]
| select(
.labels.job == "who-need-help-web" and
.labels.run_id == $run_id
)
] as $targets
| ($targets | length) == $expected
and all($targets[]; .health == "up")
] as $web_targets
| [
.data.activeTargets[]
| select(
.labels.job == "who-need-help-worker" and
.labels.run_id == $run_id
)
] as $worker_targets
| ($web_targets | length) == $expected_web
and ($worker_targets | length) == $expected_worker
and all($web_targets[]; .health == "up")
and all($worker_targets[]; .health == "up")
' "$output_dir/targets-current.json" >/dev/null; then
return 0
fi
@ -240,7 +255,7 @@ wait_for_targets_up() {
sleep 1
done
echo "Timed out waiting for every direct web metrics target." >&2
echo "Timed out waiting for every direct web and worker metrics target." >&2
return 1
}
@ -349,12 +364,18 @@ cleanup() {
trap cleanup EXIT HUP INT TERM
mapfile -t web_ids < <(service_ids web)
mapfile -t worker_ids < <(service_ids worker)
if [[ "${#web_ids[@]}" -ne "$LOAD_WEB_REPLICAS" ]]; then
echo "Expected $LOAD_WEB_REPLICAS running load web replicas; observed ${#web_ids[@]}." >&2
exit 1
fi
if [[ "${#worker_ids[@]}" -ne "$LOAD_WORKER_REPLICAS" ]]; then
echo "Expected $LOAD_WORKER_REPLICAS running load worker replicas; observed ${#worker_ids[@]}." >&2
exit 1
fi
internal_network=$(
docker network ls \
--filter "label=com.docker.compose.project=$LOAD_PROJECT" \
@ -404,6 +425,43 @@ done
jq -s '.' "$target_lines" >"$prometheus_runtime/web-targets.json"
unlink "$target_lines"
target_lines="$prometheus_runtime/worker-targets.jsonl"
: >"$target_lines"
for worker_id in "${worker_ids[@]}"; do
assert_scope "$worker_id" worker
wait_for_web "$worker_id"
worker_name=$(docker inspect --format '{{.Name}}' "$worker_id")
worker_name=${worker_name#/}
worker_ip=$(
docker inspect \
--format "{{with index .NetworkSettings.Networks \"$internal_network\"}}{{.IPAddress}}{{end}}" \
"$worker_id"
)
if [[ ! "$worker_ip" =~ ^[0-9]+([.][0-9]+){3}$ ]]; then
echo "No observed IPv4 address for $worker_name on $internal_network." >&2
exit 1
fi
jq -cn \
--arg target "$worker_ip:4000" \
--arg instance "$worker_name" \
--arg compose_project "$LOAD_PROJECT" \
--arg run_id "$LABEL" \
'{
targets: [$target],
labels: {
instance: $instance,
compose_project: $compose_project,
run_id: $run_id
}
}' >>"$target_lines"
done
jq -s '.' "$target_lines" >"$prometheus_runtime/worker-targets.json"
unlink "$target_lines"
sed \
-e "s/__SCRAPE_INTERVAL__/$OBSERVABILITY_SCRAPE_INTERVAL/g" \
-e "s/__EVALUATION_INTERVAL__/$OBSERVABILITY_EVALUATION_INTERVAL/g" \
@ -413,11 +471,14 @@ printf '%s' "$METRICS_TOKEN" >"$prometheus_runtime/metrics-token"
printf '%s' "$OBSERVABILITY_GRAFANA_ADMIN_PASSWORD" \
>"$grafana_runtime/admin-password"
cp "$prometheus_runtime/web-targets.json" "$output_dir/generated-targets.json"
cp "$prometheus_runtime/worker-targets.json" "$output_dir/generated-worker-targets.json"
chmod 600 "$prometheus_runtime/prometheus.yml" \
"$prometheus_runtime/web-targets.json" \
"$prometheus_runtime/worker-targets.json" \
"$prometheus_runtime/metrics-token" \
"$grafana_runtime/admin-password" \
"$output_dir/generated-targets.json"
"$output_dir/generated-targets.json" \
"$output_dir/generated-worker-targets.json"
set_runtime_owner "$prometheus_runtime" "65534:65534"
set_runtime_owner "$grafana_runtime" "472:0"
@ -482,33 +543,42 @@ curl --fail --silent --show-error \
if ! jq -e '
.dashboard.uid == "wnh-overview" and
.meta.provisioned == true and
(.dashboard.panels | length) == 4
(.dashboard.panels | length) == 10
' "$output_dir/grafana-dashboard.json" >/dev/null; then
echo "The provisioned Grafana dashboard did not match the tracked dashboard." >&2
exit 1
fi
wait_for_targets_up "$LOAD_WEB_REPLICAS"
wait_for_targets_up "$LOAD_WEB_REPLICAS" "$LOAD_WORKER_REPLICAS"
mv "$output_dir/targets-current.json" "$output_dir/targets-before-drill.json"
jq -n \
--slurpfile expected "$output_dir/generated-targets.json" \
--slurpfile expected_web "$output_dir/generated-targets.json" \
--slurpfile expected_worker "$output_dir/generated-worker-targets.json" \
--slurpfile observed "$output_dir/targets-before-drill.json" '
{
expected_instances: ($expected[0] | map(.labels.instance) | sort),
observed_instances: (
expected_web_instances: ($expected_web[0] | map(.labels.instance) | sort),
observed_web_instances: (
$observed[0].data.activeTargets
| map(select(.labels.job == "who-need-help-web") | .labels.instance)
| sort
),
expected_worker_instances: ($expected_worker[0] | map(.labels.instance) | sort),
observed_worker_instances: (
$observed[0].data.activeTargets
| map(select(.labels.job == "who-need-help-worker") | .labels.instance)
| sort
)
}
| . + {
exact_instance_match: (.expected_instances == .observed_instances)
exact_instance_match:
(.expected_web_instances == .observed_web_instances and
.expected_worker_instances == .observed_worker_instances)
}
' >"$output_dir/target-summary.json"
if ! jq -e '.exact_instance_match' "$output_dir/target-summary.json" >/dev/null; then
echo "Prometheus did not preserve every direct web instance target." >&2
echo "Prometheus did not preserve every direct web and worker instance target." >&2
exit 1
fi
@ -524,7 +594,7 @@ wait_for_webhook_status \
docker start "$drill_web_id" >"$output_dir/started-web.txt"
wait_for_web "$drill_web_id"
wait_for_targets_up "$LOAD_WEB_REPLICAS"
wait_for_targets_up "$LOAD_WEB_REPLICAS" "$LOAD_WORKER_REPLICAS"
mv "$output_dir/targets-current.json" "$output_dir/targets-after-recovery.json"
wait_for_alert_clear "$drill_instance"
wait_for_webhook_status \
@ -550,9 +620,11 @@ jq -n \
--arg alertmanager_url "$alertmanager_url" \
--arg grafana_url "$grafana_url" \
--argjson web_replicas "$LOAD_WEB_REPLICAS" \
--argjson worker_replicas "$LOAD_WORKER_REPLICAS" \
'{
run_id: $run_id,
web_replicas: $web_replicas,
worker_replicas: $worker_replicas,
direct_targets_up_before_and_after: true,
induced_instance: $drill_instance,
firing_webhook_observed: true,

121
scripts/quality.sh
View File

@ -22,6 +22,9 @@ backup_image="who-need-help:backup-audit-$run_id"
minio_image="who-need-help:minio-audit-$run_id"
mc_image="who-need-help:mc-audit-$run_id"
boundary_mock_image="who-need-help:boundary-mock-audit-$run_id"
socket_proxy_image="who-need-help:socket-proxy-audit-$run_id"
postgis_image="who-need-help:postgis-audit-$run_id"
socket_proxy_container="wnh-socket-proxy-audit-$run_id"
scan_dir=$(mktemp -d "${TMPDIR:-/tmp}/wnh-quality-scan.XXXXXX")
scan_list="${scan_dir}.files"
scan_tar="${scan_dir}.tar"
@ -30,14 +33,17 @@ umask 077
QUALITY_POSTGRES_USER="wnh_quality_$(openssl rand -hex 6)"
QUALITY_POSTGRES_PASSWORD=$(openssl rand -base64 48 | tr -d '\n')
export QUALITY_POSTGRES_USER QUALITY_POSTGRES_PASSWORD
QUALITY_POSTGIS_IMAGE=$postgis_image
export QUALITY_POSTGIS_IMAGE
compose="docker compose -p $project -f $ROOT/compose.quality.yaml"
cleanup() {
$compose down --volumes --remove-orphans >/dev/null 2>&1 || true
docker rm --force "$socket_proxy_container" >/dev/null 2>&1 || true
docker image rm "$quality_image" "$assets_image" "$e2e_image" "$release_image" \
"$backup_image" "$minio_image" "$mc_image" \
"$boundary_mock_image" \
"$boundary_mock_image" "$socket_proxy_image" "$postgis_image" \
>/dev/null 2>&1 || true
rm -rf "$scan_dir" "$scan_list" "$scan_tar"
}
@ -56,6 +62,7 @@ docker run --rm \
echo "Checking Dockerfiles with Hadolint 2.14.0"
for dockerfile in Dockerfile Dockerfile.backup Dockerfile.minio \
Dockerfile.postgis Dockerfile.socket-proxy \
android/Dockerfile e2e/Dockerfile ops/external-boundaries/Dockerfile; do
docker run --rm --interactive "$HADOLINT_IMAGE" \
hadolint --failure-threshold warning - <"$dockerfile"
@ -71,9 +78,42 @@ echo "Rendering every Docker Compose profile"
docker compose --env-file .env.example -f compose.yaml config --quiet
docker compose --env-file .env.example -f compose.yaml config --format json |
jq --exit-status '
[.services.migrate, .services.web, .services.worker]
| all(.environment.ERL_ZFLAGS == "+Q 65536")
. as $root
| [$root.services.migrate, $root.services.web, $root.services.worker]
| all(
.environment.ERL_ZFLAGS == "+Q 65536" and
.read_only == true and
.cap_drop == ["ALL"] and
.security_opt == ["no-new-privileges:true"] and
(.tmpfs | index("/tmp") != null)
)
and $root.services.web.environment.POOL_SIZE == "4"
and $root.services.worker.environment.POOL_SIZE == "2"
and $root.services.migrate.environment.POOL_SIZE == "2"
and $root.services.web.deploy.replicas == 2
and $root.services.worker.deploy.replicas == 2
and ($root.services.proxy.networks | keys | sort) == ["docker-api", "edge", "ingress"]
and ($root.services.web.networks | keys | sort) == ["egress", "ingress", "internal"]
and ($root.services.worker.networks | keys | sort) == ["egress", "internal"]
and ($root.services.db.networks | keys) == ["internal"]
and $root.networks.ingress.internal == true
and $root.networks.internal.internal == true
and ($root.networks.egress.internal // false) == false
and $root.services.db.security_opt == ["no-new-privileges:true"]
and $root.services.mailpit.ports[0].host_ip == "127.0.0.1"
' >/dev/null
WEB_REPLICAS=1 WORKER_REPLICAS=1 \
docker compose --env-file .env.example -f compose.yaml config --format json |
jq --exit-status '
.services.web.deploy.replicas == 1 and
.services.worker.deploy.replicas == 1
' >/dev/null
CPU_REPLAY_WEB_CPUS=1 \
CPU_REPLAY_WORKER_CPUS=1 \
CPU_REPLAY_WEB_SCHEDULERS=1 \
CPU_REPLAY_WORKER_SCHEDULERS=1 \
docker compose --env-file .env.example \
-f compose.yaml -f compose.cpu-replay.yaml config --quiet
PORTABILITY_IMAGE=who-need-help:portability-render \
docker compose --env-file .env.example \
-f compose.yaml -f compose.portability.yaml config --quiet
@ -129,6 +169,8 @@ printf '%s' 'isolated-quality-metrics-token' \
>"$scan_dir/observability-runtime/prometheus/metrics-token"
printf '%s\n' '[]' \
>"$scan_dir/observability-runtime/prometheus/web-targets.json"
printf '%s\n' '[]' \
>"$scan_dir/observability-runtime/prometheus/worker-targets.json"
docker run --rm \
--user 0:0 \
--volume "$scan_dir/observability-runtime/prometheus:/runtime:ro" \
@ -149,7 +191,7 @@ docker run --rm \
"$PYTHON_IMAGE" python -c \
'import py_compile; py_compile.compile("/src/mock_server.py", cfile="/tmp/mock_server.pyc", doraise=True)'
jq --exit-status \
'type == "object" and .uid == "wnh-overview" and (.panels | length) == 4' \
'type == "object" and .uid == "wnh-overview" and (.panels | length) == 10' \
ops/observability/grafana/dashboards/who-need-help-overview.json \
>/dev/null
@ -183,6 +225,77 @@ docker run --rm \
--exit-code 1 \
/scan
echo "Building, smoke-testing, and scanning pinned runtime infrastructure images"
docker build --tag "$socket_proxy_image" --file Dockerfile.socket-proxy .
docker build --tag "$postgis_image" --file Dockerfile.postgis .
test "$(docker image inspect --format '{{.Config.User}}' "$socket_proxy_image")" = "haproxy"
test "$(docker image inspect --format '{{.Config.User}}' "$postgis_image")" = "postgres"
docker run --rm --entrypoint sh "$postgis_image" -euc '
test ! -e /usr/local/bin/gosu
test "$(id -u)" = 70
'
docker run --detach \
--name "$socket_proxy_container" \
--read-only \
--tmpfs /run:uid=99,gid=99,mode=0755 \
--tmpfs /tmp \
--cap-drop ALL \
--group-add "$(stat -c '%g' /var/run/docker.sock)" \
--security-opt no-new-privileges \
--env CONTAINERS=1 \
--env EVENTS=1 \
--env INFO=1 \
--env NETWORKS=1 \
--env PING=1 \
--env POST=0 \
--env VERSION=1 \
--volume /var/run/docker.sock:/var/run/docker.sock:ro \
--publish 127.0.0.1::2375 \
"$socket_proxy_image" >/dev/null
socket_proxy_port=$(
docker port "$socket_proxy_container" 2375/tcp |
sed -n 's/.*://p' |
head -n 1
)
test -n "$socket_proxy_port"
socket_proxy_ready=false
for _attempt in $(seq 1 30); do
if curl --fail --silent --show-error \
"http://127.0.0.1:$socket_proxy_port/_ping" >/dev/null; then
socket_proxy_ready=true
break
fi
sleep 1
done
test "$socket_proxy_ready" = true
test "$(
curl --silent --output /dev/null --write-out '%{http_code}' \
"http://127.0.0.1:$socket_proxy_port/containers/json"
)" = "200"
test "$(
curl --silent --output /dev/null --write-out '%{http_code}' \
--request POST \
"http://127.0.0.1:$socket_proxy_port/containers/create"
)" = "403"
docker rm --force "$socket_proxy_container" >/dev/null
for image in \
"$socket_proxy_image" \
"$postgis_image" \
"traefik:v3.7.8@sha256:4299bbed850421258fc5448c2e0e6ad350981d4d335a68de11b92448aedbefe5" \
"axllent/mailpit:v1.30.4@sha256:5a49a77c5bdbe7c5474450b4f46348d09949df3695257729c93a30369382d4f6"; do
docker run --rm \
--volume /var/run/docker.sock:/var/run/docker.sock \
--volume "$ROOT/.tools/trivy-cache:/root/.cache/trivy" \
"$TRIVY_IMAGE" image \
--scanners vuln \
--severity HIGH,CRITICAL \
--ignore-unfixed \
--exit-code 1 \
"$image"
done
echo "Building the pinned quality image and cached Dialyzer PLTs"
docker build --target quality --tag "$quality_image" .

4
scripts/upgrade-rehearsal-compose.sh
View File

@ -106,6 +106,8 @@ fi
export DATABASE_URL="$database_prefix/$rehearsal_db$database_query"
export HTTP_PORT=0
export MAILPIT_PORT=0
export DOCKER_SOCKET_GID
DOCKER_SOCKET_GID=$(stat -c '%g' /var/run/docker.sock)
export PHX_HOST="$rehearsal_phx_host"
export PHX_SCHEME="$rehearsal_phx_scheme"
export PHX_URL_PORT="$rehearsal_phx_url_port"
@ -113,7 +115,7 @@ export MAP_TILE_URL="$rehearsal_map_tile_url"
export TRAEFIK_TRUSTED_IPS=127.0.0.1/32
export TRAEFIK_PROJECT_CONSTRAINT="$project"
export TRAEFIK_APP_NAME="wnh-upgrade-rehearsal-$compact_id"
export TRAEFIK_DOCKER_NETWORK="${project}_internal"
export TRAEFIK_DOCKER_NETWORK="${project}_ingress"
export REHEARSAL_IMAGE="who-need-help:upgrade-rehearsal-$run_id"
compose() {

23
test/who_need_help/accounts_test.exs
View File

@ -48,6 +48,29 @@ defmodule WhoNeedHelp.AccountsTest do
end
end
describe "public_user_query/1" do
test "does not load authentication or moderation fields" do
user =
user_fixture()
|> set_password()
|> Ecto.Changeset.change(moderation_note: "private moderator note")
|> Repo.update!()
public_user =
Accounts.public_user_query()
|> where([candidate], candidate.id == ^user.id)
|> Repo.one!()
assert public_user.id == user.id
assert public_user.display_name == user.display_name
assert is_nil(public_user.email)
assert is_nil(public_user.hashed_password)
assert is_nil(public_user.confirmed_at)
assert is_nil(public_user.accepted_terms_at)
assert is_nil(public_user.moderation_note)
end
end
describe "register_user/1" do
test "requires email to be set" do
{:error, changeset} = Accounts.register_user(%{})

5
test/who_need_help/activities_test.exs
View File

@ -196,6 +196,11 @@ defmodule WhoNeedHelp.ActivitiesTest do
assert length(Enum.uniq(actual_ids)) == 4
assert is_binary(first.next_cursor)
assert second.next_cursor == nil
assert Enum.all?(first.entries ++ second.entries, &(&1.approved_participant_count == 1))
assert Enum.all?(first.entries ++ second.entries, fn activity ->
match?(%Ecto.Association.NotLoaded{}, activity.participants)
end)
end
test "my activities are derived from the organizer and participant membership rows", context do

192
test/who_need_help/mutual_aid_flow_test.exs
View File

@ -6,6 +6,7 @@ defmodule WhoNeedHelp.MutualAidFlowTest do
alias WhoNeedHelp.{Accounts, Catalog, CatalogModeration, Help, Messaging, Tracking, Trust}
alias WhoNeedHelp.Repo
alias WhoNeedHelp.Tracking.Position
alias WhoNeedHelpWeb.Presence
setup do
category = Catalog.seed_defaults()
@ -56,6 +57,20 @@ defmodule WhoNeedHelp.MutualAidFlowTest do
assert reputation.completed == 1
assert reputation.unique_people == 1
assert reputation.verified_handovers == 1
reputations = Trust.reputations([context.helper.id, context.requester.id])
assert reputations[context.helper.id] == reputation
assert reputations[context.requester.id].completed == 1
unused_user = user_fixture(display_name: "No completed help")
assert Trust.reputations([unused_user.id])[unused_user.id] == %{
completed: 0,
unique_people: 0,
verified_handovers: 0,
location_supported: 0,
rating: nil
}
end
test "chat is durable and only visible to match participants", context do
@ -199,6 +214,47 @@ defmodule WhoNeedHelp.MutualAidFlowTest do
refute Repo.get(Position, position.id)
end
test "concurrent tracking stops serialize and broadcast once", context do
{:ok, request} = Help.create_request(context.requester_scope, context.request_attrs)
{:ok, assignment} = Help.accept_request(context.helper_scope, request.id)
{:ok, _session} = Tracking.start_session(context.helper_scope, assignment)
:ok = Tracking.subscribe(assignment.id)
results =
1..2
|> Task.async_stream(
fn _ -> Tracking.stop_session(context.helper_scope, assignment) end,
max_concurrency: 2,
ordered: false
)
|> Enum.map(fn {:ok, result} -> result end)
|> Enum.sort()
assert results == [{:ok, :already_stopped}, {:ok, :stopped}]
assert_receive {:tracking_stopped, helper_id}, 1_000
assert helper_id == context.helper.id
refute_receive {:tracking_stopped, ^helper_id}, 100
end
test "stale browser cleanup cannot stop a replacement tracking session", context do
{:ok, request} = Help.create_request(context.requester_scope, context.request_attrs)
{:ok, assignment} = Help.accept_request(context.helper_scope, request.id)
{:ok, old_session} = Tracking.start_session(context.helper_scope, assignment)
assert {:ok, :stopped} = Tracking.stop_session(context.helper_scope, assignment)
{:ok, current_session} = Tracking.start_session(context.helper_scope, assignment)
refute current_session.id == old_session.id
assert {:ok, :already_stopped} =
Tracking.stop_browser_session(
assignment.id,
context.helper.id,
old_session.id
)
assert Tracking.active_session?(context.helper_scope, assignment)
end
test "reviews remain hidden until both participants submit", context do
{:ok, request} = Help.create_request(context.requester_scope, context.request_attrs)
{:ok, assignment} = Help.accept_request(context.helper_scope, request.id)
@ -239,6 +295,15 @@ defmodule WhoNeedHelp.MutualAidFlowTest do
assert proposal_export["mode"] == "help"
assert proposal_export["community_votes"] == 0
[listed] = Catalog.list_proposals()
assert listed.id == proposal.id
assert listed.vote_count == 0
assert %Ecto.Association.NotLoaded{} = listed.votes
assert {:ok, _vote} = Catalog.vote(context.helper_scope, proposal.id)
assert [%{id: listed_id, vote_count: 1}] = Catalog.list_proposals()
assert listed_id == proposal.id
refute export =~ context.requester.email
refute export =~ context.requester.display_name
end
@ -331,4 +396,131 @@ defmodule WhoNeedHelp.MutualAidFlowTest do
assert {:error, :not_found} = Accounts.delete_social_identity(context.helper, identity.id)
assert {:ok, _identity} = Accounts.delete_social_identity(context.requester, identity.id)
end
test "request details batch public users and public social identities", context do
{:ok, requester_identity} =
Accounts.upsert_verified_social_identity(context.requester, %{
provider: :github,
provider_uid: "requester-private-provider-id",
profile_url: "https://github.com/requester",
handle: "requester",
verified_at: DateTime.utc_now(:second)
})
{:ok, helper_identity} =
Accounts.upsert_verified_social_identity(context.helper, %{
provider: :github,
provider_uid: "helper-private-provider-id",
profile_url: "https://github.com/helper",
handle: "helper",
verified_at: DateTime.utc_now(:second)
})
{:ok, request} = Help.create_request(context.requester_scope, context.request_attrs)
{:ok, _assignment} = Help.accept_request(context.helper_scope, request.id)
handler_id = "request-detail-queries-#{System.unique_integer([:positive])}"
:ok =
:telemetry.attach(
handler_id,
[:who_need_help, :repo, :query],
fn _event, _measurements, metadata, receiver ->
send(receiver, {:request_detail_query, metadata.query})
end,
self()
)
loaded =
try do
{:ok, loaded} = Help.get_request(context.requester_scope, request.id)
loaded
after
:telemetry.detach(handler_id)
end
queries = collect_request_detail_queries()
assert Enum.count(queries, &String.contains?(&1, ~s(FROM "users"))) == 1
assert Enum.count(queries, &String.contains?(&1, ~s(FROM "social_identities"))) == 1
assert [loaded_requester_identity] = loaded.requester.social_identities
assert loaded_requester_identity.id == requester_identity.id
assert is_nil(loaded_requester_identity.provider_uid)
assert [loaded_helper_identity] = loaded.assignment.helper.social_identities
assert loaded_helper_identity.id == helper_identity.id
assert is_nil(loaded_helper_identity.provider_uid)
end
test "browser tracking stops only after the last tab leaves", context do
{:ok, request} = Help.create_request(context.requester_scope, context.request_attrs)
{:ok, assignment} = Help.accept_request(context.helper_scope, request.id)
{:ok, session} = Tracking.start_session(context.helper_scope, assignment)
:ok = Tracking.subscribe(assignment.id)
parent = self()
start_browser = fn ->
spawn(fn ->
{:ok, key} =
Presence.track_browser(self(), assignment.id, context.helper.id, session.id)
send(parent, {:browser_ready, self(), key})
receive do
:close -> :ok
end
end)
end
first = start_browser.()
second = start_browser.()
key = Presence.tracking_key(assignment.id, context.helper.id)
assert_receive {:browser_ready, ^first, ^key}
assert_receive {:browser_ready, ^second, ^key}
assert wait_for_presence_count(key, 2)
first_ref = Process.monitor(first)
send(first, :close)
assert_receive {:DOWN, ^first_ref, :process, ^first, :normal}
assert wait_for_presence_count(key, 1)
assert Tracking.active_session?(context.helper_scope, assignment)
second_ref = Process.monitor(second)
send(second, :close)
assert_receive {:DOWN, ^second_ref, :process, ^second, :normal}
assert_receive {:tracking_stopped, helper_id}, 1_000
assert helper_id == context.helper.id
refute Tracking.active_session?(context.helper_scope, assignment)
end
defp wait_for_presence_count(key, expected, attempts \\ 100)
defp wait_for_presence_count(_key, _expected, 0), do: false
defp wait_for_presence_count(key, expected, attempts) do
count =
case Presence.get_by_key(Presence.tracking_topic(), key) do
[] -> 0
%{metas: metas} -> length(metas)
end
if count == expected do
true
else
Process.sleep(10)
wait_for_presence_count(key, expected, attempts - 1)
end
end
defp collect_request_detail_queries(queries \\ []) do
receive do
{:request_detail_query, query} ->
collect_request_detail_queries([query | queries])
after
0 -> Enum.reverse(queries)
end
end
end

65
test/who_need_help/workers/expire_requests_test.exs Normal file
View File

@ -0,0 +1,65 @@
defmodule WhoNeedHelp.Workers.ExpireRequestsTest do
use WhoNeedHelp.DataCase, async: false
import WhoNeedHelp.AccountsFixtures
alias WhoNeedHelp.{Catalog, Help, Repo}
alias WhoNeedHelp.Help.HelpRequest
alias WhoNeedHelp.Trust.AuditEvent
alias WhoNeedHelp.Workers.ExpireRequests
test "expires an arbitrary backlog without loading it as one result set" do
category = Catalog.seed_defaults()
requester = user_fixture(display_name: "Expiration requester")
scope = user_scope_fixture(requester)
request_ids =
for number <- 1..12 do
{:ok, request} =
Help.create_request(scope, %{
"title" => "Expiring medicine request #{number}",
"description" => "The reserved legal medicine was not collected in time.",
"pickup_instructions" => "No payment or prescription handling is required.",
"location_label" => "Central district",
"latitude" => "50.4501",
"longitude" => "30.5234",
"urgency" => "today",
"location_visibility" => "approximate_public",
"structured_data" => %{"pickup_status" => "reserved"},
"expires_at" => DateTime.utc_now(:second) |> DateTime.add(3, :hour),
"category_id" => category.id
})
request.id
end
past = DateTime.utc_now(:second) |> DateTime.add(-1, :minute)
Repo.update_all(
from(request in HelpRequest, where: request.id in ^request_ids),
set: [expires_at: past]
)
assert {:ok, %{expired: 12}} = ExpireRequests.perform(%Oban.Job{})
assert Repo.aggregate(
from(
request in HelpRequest,
where: request.id in ^request_ids and request.status == :expired
),
:count
) == 12
assert Repo.aggregate(
from(
event in AuditEvent,
where:
event.action == "request.expired" and
event.target_id in ^request_ids
),
:count
) == 12
assert {:ok, %{expired: 0}} = ExpireRequests.perform(%Oban.Job{})
end
end

37
test/who_need_help_web/worker_metrics_plug_test.exs Normal file
View File

@ -0,0 +1,37 @@
defmodule WhoNeedHelpWeb.WorkerMetricsPlugTest do
use WhoNeedHelp.DataCase, async: false
import Plug.Conn
import Plug.Test
alias WhoNeedHelpWeb.WorkerMetricsPlug
test "liveness and readiness are available without exposing metrics" do
live = WorkerMetricsPlug.call(conn(:get, "/healthz/live"), [])
ready = WorkerMetricsPlug.call(conn(:get, "/healthz/ready"), [])
assert live.status == 200
assert ready.status == 200
assert Jason.decode!(live.resp_body)["role"] == "worker"
assert Jason.decode!(ready.resp_body)["status"] == "ready"
end
test "metrics require the independent bearer token" do
unauthorized = WorkerMetricsPlug.call(conn(:get, "/metrics"), [])
authorized =
conn(:get, "/metrics")
|> put_req_header("authorization", "Bearer test-metrics-token")
|> WorkerMetricsPlug.call([])
assert unauthorized.status == 401
assert get_resp_header(unauthorized, "www-authenticate") == ["Bearer"]
assert authorized.status == 200
assert authorized.resp_body =~ "who_need_help_database_query_duration_seconds"
end
test "all other methods and paths are unavailable" do
assert WorkerMetricsPlug.call(conn(:post, "/metrics"), []).status == 404
assert WorkerMetricsPlug.call(conn(:get, "/"), []).status == 404
end
end