diff --git a/.dialyzer_ignore.exs b/.dialyzer_ignore.exs
index c01bce1..4661c98 100644
--- a/.dialyzer_ignore.exs
+++ b/.dialyzer_ignore.exs
@@ -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}
]
diff --git a/.env.e2e.example b/.env.e2e.example
index b5de97b..e42a29e 100644
--- a/.env.e2e.example
+++ b/.env.e2e.example
@@ -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
diff --git a/.env.example b/.env.example
index d7f4c7f..ee2811b 100644
--- a/.env.example
+++ b/.env.example
@@ -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.
diff --git a/.env.load.example b/.env.load.example
index 1dd5f37..d402f29 100644
--- a/.env.load.example
+++ b/.env.load.example
@@ -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
diff --git a/Dockerfile.postgis b/Dockerfile.postgis
new file mode 100644
index 0000000..0109a66
--- /dev/null
+++ b/Dockerfile.postgis
@@ -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
diff --git a/Dockerfile.socket-proxy b/Dockerfile.socket-proxy
new file mode 100644
index 0000000..30088c9
--- /dev/null
+++ b/Dockerfile.socket-proxy
@@ -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
diff --git a/compose.cpu-replay.yaml b/compose.cpu-replay.yaml
new file mode 100644
index 0000000..6c60c2c
--- /dev/null
+++ b/compose.cpu-replay.yaml
@@ -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}"
diff --git a/compose.load.yaml b/compose.load.yaml
index 62400e8..1381d82 100644
--- a/compose.load.yaml
+++ b/compose.load.yaml
@@ -15,7 +15,7 @@ services:
edge:
aliases:
- ${LOAD_HOST}
- internal: {}
+ ingress: {}
migrate:
image: who-need-help:load
diff --git a/compose.quality.yaml b/compose.quality.yaml
index 7ce7a73..0e8120c 100644
--- a/compose.quality.yaml
+++ b/compose.quality.yaml
@@ -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}
diff --git a/compose.yaml b/compose.yaml
index 1ab0648..f3c6627 100644
--- a/compose.yaml
+++ b/compose.yaml
@@ -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:
diff --git a/config/runtime.exs b/config/runtime.exs
index de75ed7..b58f412 100644
--- a/config/runtime.exs
+++ b/config/runtime.exs
@@ -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
diff --git a/deploy/helm/who-need-help/templates/deployments.yaml b/deploy/helm/who-need-help/templates/deployments.yaml
index d147fca..33b4ba3 100644
--- a/deploy/helm/who-need-help/templates/deployments.yaml
+++ b/deploy/helm/who-need-help/templates/deployments.yaml
@@ -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
diff --git a/deploy/helm/who-need-help/templates/migrate-job.yaml b/deploy/helm/who-need-help/templates/migrate-job.yaml
index 9a5493c..80c4db3 100644
--- a/deploy/helm/who-need-help/templates/migrate-job.yaml
+++ b/deploy/helm/who-need-help/templates/migrate-job.yaml
@@ -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
diff --git a/deploy/helm/who-need-help/templates/services.yaml b/deploy/helm/who-need-help/templates/services.yaml
index 856c1d8..457e0b7 100644
--- a/deploy/helm/who-need-help/templates/services.yaml
+++ b/deploy/helm/who-need-help/templates/services.yaml
@@ -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:
diff --git a/deploy/helm/who-need-help/values.yaml b/deploy/helm/who-need-help/values.yaml
index 03f6e98..4b91478 100644
--- a/deploy/helm/who-need-help/values.yaml
+++ b/deploy/helm/who-need-help/values.yaml
@@ -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.
diff --git a/docs/performance.md b/docs/performance.md
index 2562606..2e9c2fa 100644
--- a/docs/performance.md
+++ b/docs/performance.md
@@ -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
diff --git a/lib/who_need_help/accounts.ex b/lib/who_need_help/accounts.ex
index 2f331a8..c5e32a8 100644
--- a/lib/who_need_help/accounts.ex
+++ b/lib/who_need_help/accounts.ex
@@ -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
diff --git a/lib/who_need_help/activities.ex b/lib/who_need_help/activities.ex
index 0206785..e3eb830 100644
--- a/lib/who_need_help/activities.ex
+++ b/lib/who_need_help/activities.ex
@@ -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
diff --git a/lib/who_need_help/activities/activity.ex b/lib/who_need_help/activities/activity.ex
index c570862..a64e3db 100644
--- a/lib/who_need_help/activities/activity.ex
+++ b/lib/who_need_help/activities/activity.ex
@@ -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
diff --git a/lib/who_need_help/application.ex b/lib/who_need_help/application.ex
index 4e78aab..8db5bf5 100644
--- a/lib/who_need_help/application.ex
+++ b/lib/who_need_help/application.ex
@@ -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
diff --git a/lib/who_need_help/catalog.ex b/lib/who_need_help/catalog.ex
index efd4626..4e0c84e 100644
--- a/lib/who_need_help/catalog.ex
+++ b/lib/who_need_help/catalog.ex
@@ -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
diff --git a/lib/who_need_help/catalog/category_proposal.ex b/lib/who_need_help/catalog/category_proposal.ex
index 318d9c7..abeaac1 100644
--- a/lib/who_need_help/catalog/category_proposal.ex
+++ b/lib/who_need_help/catalog/category_proposal.ex
@@ -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)
diff --git a/lib/who_need_help/catalog_moderation.ex b/lib/who_need_help/catalog_moderation.ex
index 8a15ac0..f9dfb14 100644
--- a/lib/who_need_help/catalog_moderation.ex
+++ b/lib/who_need_help/catalog_moderation.ex
@@ -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)
diff --git a/lib/who_need_help/help.ex b/lib/who_need_help/help.ex
index ed56737..97da558 100644
--- a/lib/who_need_help/help.ex
+++ b/lib/who_need_help/help.ex
@@ -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
diff --git a/lib/who_need_help/messaging.ex b/lib/who_need_help/messaging.ex
index 2f25ef6..c878a7e 100644
--- a/lib/who_need_help/messaging.ex
+++ b/lib/who_need_help/messaging.ex
@@ -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,
diff --git a/lib/who_need_help/tracking.ex b/lib/who_need_help/tracking.ex
index 70f5e0c..5bc5427 100644
--- a/lib/who_need_help/tracking.ex
+++ b/lib/who_need_help/tracking.ex
@@ -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_helper_movement(assignment, user.id, session.movement_observed_at, now)
+ 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
+ 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
+
+ :ok
end
- defp maybe_mark_proximity(%Assignment{proximity_observed_at: observed} = assignment, _, _, _)
+ 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
diff --git a/lib/who_need_help/tracking_presence_cleanup.ex b/lib/who_need_help/tracking_presence_cleanup.ex
new file mode 100644
index 0000000..22f066c
--- /dev/null
+++ b/lib/who_need_help/tracking_presence_cleanup.ex
@@ -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
diff --git a/lib/who_need_help/trust.ex b/lib/who_need_help/trust.ex
index 2bed1e7..44ab858 100644
--- a/lib/who_need_help/trust.ex
+++ b/lib/who_need_help/trust.ex
@@ -25,21 +25,17 @@ defmodule WhoNeedHelp.Trust do
}
def authorize_action(%Scope{user: %User{} = user}, action) do
- user = Repo.get(User, user.id)
+ 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)
+ error
- cond do
- not Accounts.eligible_for_trust_actions?(user) ->
- {:error, :account_not_eligible}
-
- true ->
- case RateLimiter.check(action, user.id) do
- {:error, :rate_limited} = error ->
- create_velocity_signal_once(user.id, action)
- error
-
- result ->
- result
- end
+ 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
diff --git a/lib/who_need_help/workers/expire_requests.ex b/lib/who_need_help/workers/expire_requests.ex
index bbb1b6c..e1c9e71 100644
--- a/lib/who_need_help/workers/expire_requests.ex
+++ b/lib/who_need_help/workers/expire_requests.ex
@@ -12,25 +12,39 @@ 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()
+ with {:ok, expired} <- expire_available(now, 0),
+ {:ok, cleanup} <- Tracking.cleanup_finished_sessions() do
+ {pruned_buckets, _} = RateLimiter.prune_expired()
- 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()
+ {:ok,
+ %{
+ expired: expired,
+ tracking_positions_deleted: cleanup.positions_deleted,
+ tracking_sessions_ended: cleanup.sessions_ended,
+ rate_limit_buckets_pruned: pruned_buckets
+ }}
+ end
+ end
- if request do
+ # 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)
@@ -39,25 +53,13 @@ defmodule WhoNeedHelp.Workers.ExpireRequests do
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()
- {pruned_buckets, _} = RateLimiter.prune_expired()
-
- {:ok,
- %{
- expired: expired,
- tracking_positions_deleted: cleanup.positions_deleted,
- tracking_sessions_ended: cleanup.sessions_ended,
- rate_limit_buckets_pruned: pruned_buckets
- }}
+ case result do
+ {:ok, :empty} -> {:ok, expired}
+ {:ok, %HelpRequest{}} -> expire_available(now, expired + 1)
+ {:error, reason} -> {:error, reason}
+ end
end
end
diff --git a/lib/who_need_help_web/controllers/metrics_controller.ex b/lib/who_need_help_web/controllers/metrics_controller.ex
index aea0839..6bf417b 100644
--- a/lib/who_need_help_web/controllers/metrics_controller.ex
+++ b/lib/who_need_help_web/controllers/metrics_controller.ex
@@ -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
diff --git a/lib/who_need_help_web/live/activity_live/index.ex b/lib/who_need_help_web/live/activity_live/index.ex
index 3e53d9c..95db499 100644
--- a/lib/who_need_help_web/live/activity_live/index.ex
+++ b/lib/who_need_help_web/live/activity_live/index.ex
@@ -175,7 +175,7 @@ defmodule WhoNeedHelpWeb.ActivityLive.Index do
)}
- {Enum.count(activity.participants, &(&1.status == :approved))}/{activity.capacity}
+ {activity.approved_participant_count}/{activity.capacity}
{activity.title}
diff --git a/lib/who_need_help_web/live/category_proposal_live.ex b/lib/who_need_help_web/live/category_proposal_live.ex
index b666f06..bfa66d4 100644
--- a/lib/who_need_help_web/live/category_proposal_live.ex
+++ b/lib/who_need_help_web/live/category_proposal_live.ex
@@ -148,7 +148,7 @@ defmodule WhoNeedHelpWeb.CategoryProposalLive do
diff --git a/lib/who_need_help_web/live/moderation_live.ex b/lib/who_need_help_web/live/moderation_live.ex
index 8ed1dd9..10362cc 100644
--- a/lib/who_need_help_web/live/moderation_live.ex
+++ b/lib/who_need_help_web/live/moderation_live.ex
@@ -478,7 +478,7 @@ defmodule WhoNeedHelpWeb.ModerationLive do
{proposal.reason}
- {gettext("%{count} community votes", count: length(proposal.votes))}
+ {gettext("%{count} community votes", count: proposal.vote_count)}
<.form
diff --git a/lib/who_need_help_web/live/request_live/show.ex b/lib/who_need_help_web/live/request_live/show.ex
index 2abdd94..c01c740 100644
--- a/lib/who_need_help_web/live/request_live/show.ex
+++ b/lib/who_need_help_web/live/request_live/show.ex
@@ -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)
diff --git a/lib/who_need_help_web/metrics_access.ex b/lib/who_need_help_web/metrics_access.ex
new file mode 100644
index 0000000..c5216c7
--- /dev/null
+++ b/lib/who_need_help_web/metrics_access.ex
@@ -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
diff --git a/lib/who_need_help_web/presence.ex b/lib/who_need_help_web/presence.ex
index c16950f..e7670dc 100644
--- a/lib/who_need_help_web/presence.ex
+++ b/lib/who_need_help_web/presence.ex
@@ -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
diff --git a/lib/who_need_help_web/telemetry.ex b/lib/who_need_help_web/telemetry.ex
index f8f8d30..eba7ced 100644
--- a/lib/who_need_help_web/telemetry.ex
+++ b/lib/who_need_help_web/telemetry.ex
@@ -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,
diff --git a/lib/who_need_help_web/worker_metrics_plug.ex b/lib/who_need_help_web/worker_metrics_plug.ex
new file mode 100644
index 0000000..6e8d39a
--- /dev/null
+++ b/lib/who_need_help_web/worker_metrics_plug.ex
@@ -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
diff --git a/ops/observability/grafana/dashboards/who-need-help-overview.json b/ops/observability/grafana/dashboards/who-need-help-overview.json
index a7fd7f0..a9ac15a 100644
--- a/ops/observability/grafana/dashboards/who-need-help-overview.json
+++ b/ops/observability/grafana/dashboards/who-need-help-overview.json
@@ -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",
diff --git a/ops/observability/prometheus.template.yml b/ops/observability/prometheus.template.yml
index 243a379..767f047 100644
--- a/ops/observability/prometheus.template.yml
+++ b/ops/observability/prometheus.template.yml
@@ -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__
diff --git a/ops/observability/rules.yml b/ops/observability/rules.yml
index 342e70d..15f63be 100644
--- a/ops/observability/rules.yml
+++ b/ops/observability/rules.yml
@@ -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."
diff --git a/scripts/clean-deploy-verify.sh b/scripts/clean-deploy-verify.sh
index e1fe757..a3d66d3 100755
--- a/scripts/clean-deploy-verify.sh
+++ b/scripts/clean-deploy-verify.sh
@@ -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" <"$target" <"$temporary"
if grep -Eq '^[A-Z0-9_]+=GENERATE_' "$temporary"; then
diff --git a/scripts/load-run.sh b/scripts/load-run.sh
index 9aa8a4e..9558e93 100755
--- a/scripts/load-run.sh
+++ b/scripts/load-run.sh
@@ -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:
diff --git a/scripts/observability-run.sh b/scripts/observability-run.sh
index b21a877..6ccb8a0 100755
--- a/scripts/observability-run.sh
+++ b/scripts/observability-run.sh
@@ -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
- exit 1
-fi
+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
+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,
diff --git a/scripts/quality.sh b/scripts/quality.sh
index cec7b96..97a9da0 100755
--- a/scripts/quality.sh
+++ b/scripts/quality.sh
@@ -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" .
diff --git a/scripts/upgrade-rehearsal-compose.sh b/scripts/upgrade-rehearsal-compose.sh
index 69ed0b3..8941282 100755
--- a/scripts/upgrade-rehearsal-compose.sh
+++ b/scripts/upgrade-rehearsal-compose.sh
@@ -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() {
diff --git a/test/who_need_help/accounts_test.exs b/test/who_need_help/accounts_test.exs
index 4673684..f74e988 100644
--- a/test/who_need_help/accounts_test.exs
+++ b/test/who_need_help/accounts_test.exs
@@ -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(%{})
diff --git a/test/who_need_help/activities_test.exs b/test/who_need_help/activities_test.exs
index 32d9747..e0007e3 100644
--- a/test/who_need_help/activities_test.exs
+++ b/test/who_need_help/activities_test.exs
@@ -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
diff --git a/test/who_need_help/mutual_aid_flow_test.exs b/test/who_need_help/mutual_aid_flow_test.exs
index 5f9e754..10f76b2 100644
--- a/test/who_need_help/mutual_aid_flow_test.exs
+++ b/test/who_need_help/mutual_aid_flow_test.exs
@@ -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
diff --git a/test/who_need_help/workers/expire_requests_test.exs b/test/who_need_help/workers/expire_requests_test.exs
new file mode 100644
index 0000000..91c238a
--- /dev/null
+++ b/test/who_need_help/workers/expire_requests_test.exs
@@ -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
diff --git a/test/who_need_help_web/worker_metrics_plug_test.exs b/test/who_need_help_web/worker_metrics_plug_test.exs
new file mode 100644
index 0000000..1f044a5
--- /dev/null
+++ b/test/who_need_help_web/worker_metrics_plug_test.exs
@@ -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