feat: deliver durable push product events

.env.example

@@ -46,6 +46,17 @@ GITHUB_OAUTH_USER_URL=
 GITHUB_OAUTH_HTTP_CONNECT_TIMEOUT_MS=
 GITHUB_OAUTH_HTTP_RECEIVE_TIMEOUT_MS=
 
+# Optional provider-neutral HTTP push boundary. Leave both endpoint and token
+# empty to disable product push jobs. When enabled, all four numeric values are
+# required deployment inputs; the project does not claim universal production
+# values for them.
+PUSH_HTTP_ENDPOINT=
+PUSH_HTTP_BEARER_TOKEN=
+PUSH_HTTP_MAX_ATTEMPTS=
+PUSH_HTTP_RECEIVE_TIMEOUT_MS=
+PUSH_HTTP_CONNECT_TIMEOUT_MS=
+PUSH_HTTP_RETRY_DELAY_MS=
+
 POSTGRES_DB=who_need_help
 POSTGRES_USER=postgres
 POSTGRES_PASSWORD=replace-with-a-local-or-deployment-secret

README.md

@@ -118,7 +118,7 @@ restores it into a new temporary database, checks corruption and interruption
 failure paths, removes those temporary buckets, and retains the successful
 encrypted bucket in local MinIO. It never writes a plaintext dump to the host.
 
-Exercise the real OAuth/SMTP protocol clients and the future push adapter
+Exercise the real OAuth/SMTP protocol clients and the provider-neutral push
 boundary entirely inside an isolated Docker network:
 
 ```bash
@@ -128,8 +128,12 @@ boundary entirely inside an isolated Docker network:
 The command publishes no host ports, generates independent one-run credentials
 in an ignored mode-`0600` file, verifies success, rejection, retry, replay, and
 timeout paths, retains only non-secret evidence, and removes its exact Compose
-project and images. The HTTP push adapter is deliberately not connected to a
+project, database volume, and images. It also runs two Oban worker replicas and
-product workflow and is not presented as an FCM or APNs implementation.
+verifies that request acceptance and new-chat events are delivered from their
+real domain transactions, including an Oban retry and pre-transport replay
+deduplication. The adapter is not presented as an FCM or APNs implementation;
+an external provider must resolve the stable user recipient to registered
+devices.
 
 The commands, boundaries, and unclaimed production properties are documented
 in [the operations runbook](docs/operations.md).

compose.external-boundaries.yaml

@@ -1,6 +1,53 @@
 name: who_need_help_external_boundaries
 
+x-boundary-app-environment: &boundary-app-environment
+  DATABASE_URL: ${EXTERNAL_DATABASE_URL:?Set EXTERNAL_DATABASE_URL}
+  SECRET_KEY_BASE: ${EXTERNAL_SECRET_KEY_BASE:?Set EXTERNAL_SECRET_KEY_BASE}
+  HANDOVER_SECRET: ${EXTERNAL_HANDOVER_SECRET:?Set EXTERNAL_HANDOVER_SECRET}
+  PHX_HOST: boundary.local
+  PHX_SCHEME: http
+  PHX_URL_PORT: "80"
+  GITHUB_OAUTH_CLIENT_ID: ${EXTERNAL_OAUTH_CLIENT_ID:?Set EXTERNAL_OAUTH_CLIENT_ID}
+  GITHUB_OAUTH_CLIENT_SECRET: ${EXTERNAL_OAUTH_CLIENT_SECRET:?Set EXTERNAL_OAUTH_CLIENT_SECRET}
+  GITHUB_OAUTH_BASE_URL: http://external-mock:8080
+  GITHUB_OAUTH_AUTHORIZE_URL: http://external-mock:8080/oauth/authorize
+  GITHUB_OAUTH_TOKEN_URL: http://external-mock:8080/oauth/token
+  GITHUB_OAUTH_USER_URL: http://external-mock:8080/oauth/user
+  GITHUB_OAUTH_HTTP_CONNECT_TIMEOUT_MS: "100"
+  GITHUB_OAUTH_HTTP_RECEIVE_TIMEOUT_MS: "100"
+  SMTP_RELAY: external-mock
+  SMTP_PORT: "2525"
+  SMTP_USERNAME: ""
+  SMTP_PASSWORD: ""
+  SMTP_AUTH: never
+  SMTP_TLS: never
+  SMTP_SSL: "false"
+  EMAIL_FROM_NAME: Who Need Help boundary
+  EMAIL_FROM_ADDRESS: boundary@example.invalid
+  PUSH_HTTP_ENDPOINT: http://external-mock:8080/push
+  PUSH_HTTP_BEARER_TOKEN: ${EXTERNAL_PUSH_BEARER_TOKEN:?Set EXTERNAL_PUSH_BEARER_TOKEN}
+  PUSH_HTTP_MAX_ATTEMPTS: "1"
+  PUSH_HTTP_RECEIVE_TIMEOUT_MS: "1000"
+  PUSH_HTTP_CONNECT_TIMEOUT_MS: "1000"
+  PUSH_HTTP_RETRY_DELAY_MS: "0"
+
 services:
+  boundary-db:
+    image: postgis/postgis:18-3.6-alpine@sha256:05d68c7f0f19b9aa0bf7c4a2049b2e8b38b44a63116392b95726a4c913766cf6
+    environment:
+      POSTGRES_DB: boundary
+      POSTGRES_USER: boundary
+      POSTGRES_PASSWORD: ${EXTERNAL_POSTGRES_PASSWORD:?Set EXTERNAL_POSTGRES_PASSWORD}
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -h 127.0.0.1 -U $$POSTGRES_USER -d $$POSTGRES_DB"]
+      interval: 1s
+      timeout: 2s
+      retries: 60
+    volumes:
+      - boundary_postgres_data:/var/lib/postgresql
+    networks: [boundary]
+    restart: "no"
+
   external-mock:
     image: ${EXTERNAL_BOUNDARY_MOCK_IMAGE:?Set EXTERNAL_BOUNDARY_MOCK_IMAGE}
     build:
@@ -30,6 +77,50 @@ services:
     networks: [boundary]
     restart: "no"
 
+  boundary-migrate:
+    image: ${EXTERNAL_BOUNDARY_APP_IMAGE:?Set EXTERNAL_BOUNDARY_APP_IMAGE}
+    build:
+      context: .
+      target: release
+    command: ["/app/bin/migrate"]
+    environment:
+      <<: *boundary-app-environment
+      APP_ROLE: migrate
+    depends_on:
+      boundary-db:
+        condition: service_healthy
+    read_only: true
+    tmpfs:
+      - /tmp
+    cap_drop: [ALL]
+    security_opt:
+      - no-new-privileges:true
+    networks: [boundary]
+    restart: "no"
+
+  boundary-worker:
+    image: ${EXTERNAL_BOUNDARY_APP_IMAGE:?Set EXTERNAL_BOUNDARY_APP_IMAGE}
+    build:
+      context: .
+      target: release
+    command: ["/app/bin/who_need_help", "start"]
+    environment:
+      <<: *boundary-app-environment
+      APP_ROLE: worker
+    depends_on:
+      boundary-migrate:
+        condition: service_completed_successfully
+      external-mock:
+        condition: service_healthy
+    read_only: true
+    tmpfs:
+      - /tmp
+    cap_drop: [ALL]
+    security_opt:
+      - no-new-privileges:true
+    networks: [boundary]
+    restart: "no"
+
   boundary-check:
     image: ${EXTERNAL_BOUNDARY_APP_IMAGE:?Set EXTERNAL_BOUNDARY_APP_IMAGE}
     build:
@@ -40,34 +131,17 @@ services:
       - eval
       - WhoNeedHelp.ExternalBoundaryDrill.run!()
     environment:
+      <<: *boundary-app-environment
       APP_ROLE: migrate
-      DATABASE_URL: ${EXTERNAL_DATABASE_URL:?Set EXTERNAL_DATABASE_URL}
-      SECRET_KEY_BASE: ${EXTERNAL_SECRET_KEY_BASE:?Set EXTERNAL_SECRET_KEY_BASE}
-      HANDOVER_SECRET: ${EXTERNAL_HANDOVER_SECRET:?Set EXTERNAL_HANDOVER_SECRET}
-      PHX_HOST: boundary.local
-      PHX_SCHEME: http
-      PHX_URL_PORT: "80"
-      GITHUB_OAUTH_CLIENT_ID: ${EXTERNAL_OAUTH_CLIENT_ID:?Set EXTERNAL_OAUTH_CLIENT_ID}
-      GITHUB_OAUTH_CLIENT_SECRET: ${EXTERNAL_OAUTH_CLIENT_SECRET:?Set EXTERNAL_OAUTH_CLIENT_SECRET}
-      GITHUB_OAUTH_BASE_URL: http://external-mock:8080
-      GITHUB_OAUTH_AUTHORIZE_URL: http://external-mock:8080/oauth/authorize
-      GITHUB_OAUTH_TOKEN_URL: http://external-mock:8080/oauth/token
-      GITHUB_OAUTH_USER_URL: http://external-mock:8080/oauth/user
-      GITHUB_OAUTH_HTTP_CONNECT_TIMEOUT_MS: "100"
-      GITHUB_OAUTH_HTTP_RECEIVE_TIMEOUT_MS: "100"
-      SMTP_RELAY: external-mock
-      SMTP_PORT: "2525"
-      SMTP_USERNAME: ""
-      SMTP_PASSWORD: ""
-      SMTP_AUTH: never
-      SMTP_TLS: never
-      SMTP_SSL: "false"
-      EMAIL_FROM_NAME: Who Need Help boundary
-      EMAIL_FROM_ADDRESS: boundary@example.invalid
       EXTERNAL_MOCK_BASE_URL: http://external-mock:8080
       EXTERNAL_SMTP_RELAY: external-mock
       EXTERNAL_PUSH_BEARER_TOKEN: ${EXTERNAL_PUSH_BEARER_TOKEN:?Set EXTERNAL_PUSH_BEARER_TOKEN}
+      EXTERNAL_EXPECTED_WORKER_REPLICAS: "2"
     depends_on:
+      boundary-migrate:
+        condition: service_completed_successfully
+      boundary-worker:
+        condition: service_started
       external-mock:
         condition: service_healthy
     volumes:
@@ -85,3 +159,6 @@ services:
 networks:
   boundary:
     internal: true
+
+volumes:
+  boundary_postgres_data:

compose.yaml

@@ -33,6 +33,12 @@ x-app-environment: &app-environment
   GITHUB_OAUTH_USER_URL: ${GITHUB_OAUTH_USER_URL:-}
   GITHUB_OAUTH_HTTP_CONNECT_TIMEOUT_MS: ${GITHUB_OAUTH_HTTP_CONNECT_TIMEOUT_MS:-}
   GITHUB_OAUTH_HTTP_RECEIVE_TIMEOUT_MS: ${GITHUB_OAUTH_HTTP_RECEIVE_TIMEOUT_MS:-}
+  PUSH_HTTP_ENDPOINT: ${PUSH_HTTP_ENDPOINT:-}
+  PUSH_HTTP_BEARER_TOKEN: ${PUSH_HTTP_BEARER_TOKEN:-}
+  PUSH_HTTP_MAX_ATTEMPTS: ${PUSH_HTTP_MAX_ATTEMPTS:-}
+  PUSH_HTTP_RECEIVE_TIMEOUT_MS: ${PUSH_HTTP_RECEIVE_TIMEOUT_MS:-}
+  PUSH_HTTP_CONNECT_TIMEOUT_MS: ${PUSH_HTTP_CONNECT_TIMEOUT_MS:-}
+  PUSH_HTTP_RETRY_DELAY_MS: ${PUSH_HTTP_RETRY_DELAY_MS:-}
 
 services:
   proxy:

config/config.exs

@@ -39,7 +39,7 @@ config :geo_postgis, json_library: Jason
 
 config :who_need_help, Oban,
   repo: WhoNeedHelp.Repo,
-  queues: [default: 10, maintenance: 2],
+  queues: [default: 10, maintenance: 2, push: 1],
   plugins: [
     {Oban.Plugins.Pruner, max_age: 86_400},
     {Oban.Plugins.Cron, crontab: [{"* * * * *", WhoNeedHelp.Workers.ExpireRequests}]}

config/runtime.exs

@@ -39,6 +39,29 @@ optional_positive_integer = fn name ->
   end
 end
 
+required_positive_integer = fn name ->
+  case optional_positive_integer.(name) do
+    nil -> raise "#{name} is required when the HTTP push boundary is enabled."
+    value -> value
+  end
+end
+
+required_non_negative_integer = fn name ->
+  case System.get_env(name) do
+    value when value in [nil, ""] ->
+      raise "#{name} is required when the HTTP push boundary is enabled."
+
+    value ->
+      case Integer.parse(value) do
+        {integer, ""} when integer >= 0 ->
+          integer
+
+        _other ->
+          raise "#{name} must be a non-negative integer."
+      end
+  end
+end
+
 oauth_endpoint = fn name, default ->
   value =
     case System.get_env(name) do
@@ -110,6 +133,52 @@ github_oauth =
 
 config :who_need_help, :social_oauth, github_oauth
 
+push_configuration =
+  case {
+    System.get_env("PUSH_HTTP_ENDPOINT"),
+    System.get_env("PUSH_HTTP_BEARER_TOKEN")
+  } do
+    {endpoint, bearer_token}
+    when is_binary(endpoint) and endpoint != "" and is_binary(bearer_token) and
+           bearer_token != "" ->
+      case URI.parse(endpoint) do
+        %URI{scheme: scheme, host: host}
+        when scheme in ["http", "https"] and is_binary(host) and host != "" ->
+          :ok
+
+        _other ->
+          raise "PUSH_HTTP_ENDPOINT must be an absolute HTTP or HTTPS URL."
+      end
+
+      [
+        adapter: WhoNeedHelp.Push.HTTPAdapter,
+        endpoint: endpoint,
+        bearer_token: bearer_token,
+        max_attempts: required_positive_integer.("PUSH_HTTP_MAX_ATTEMPTS"),
+        receive_timeout: required_positive_integer.("PUSH_HTTP_RECEIVE_TIMEOUT_MS"),
+        connect_timeout: required_positive_integer.("PUSH_HTTP_CONNECT_TIMEOUT_MS"),
+        retry_delay_ms: required_non_negative_integer.("PUSH_HTTP_RETRY_DELAY_MS")
+      ]
+
+    {endpoint, bearer_token} when endpoint in [nil, ""] and bearer_token in [nil, ""] ->
+      []
+
+    _partial_configuration ->
+      raise """
+      PUSH_HTTP_ENDPOINT and PUSH_HTTP_BEARER_TOKEN must either both be set or both be empty.
+      """
+  end
+
+config :who_need_help,
+  push_product_enabled: push_configuration != [],
+  push_adapter:
+    Keyword.get(
+      push_configuration,
+      :adapter,
+      WhoNeedHelp.Push.DisabledAdapter
+    ),
+  push_delivery_options: Keyword.delete(push_configuration, :adapter)
+
 if config_env() == :prod and app_role == :web do
   metrics_token =
     System.get_env("METRICS_TOKEN") ||

deploy/helm/who-need-help/values.yaml

@@ -30,7 +30,10 @@ app:
 # Required. The Secret must contain DATABASE_URL, SECRET_KEY_BASE,
 # HANDOVER_SECRET, RELEASE_COOKIE, and METRICS_TOKEN. It may also contain both
 # GITHUB_OAUTH_CLIENT_ID and GITHUB_OAUTH_CLIENT_SECRET to enable verified
-# GitHub linking.
+# GitHub linking. Push is opt-in: provide PUSH_HTTP_ENDPOINT,
+# PUSH_HTTP_BEARER_TOKEN, PUSH_HTTP_MAX_ATTEMPTS,
+# PUSH_HTTP_RECEIVE_TIMEOUT_MS, PUSH_HTTP_CONNECT_TIMEOUT_MS, and
+# PUSH_HTTP_RETRY_DELAY_MS as one complete set.
 existingSecret: ""
 
 ingress:

docs/architecture.md

@@ -8,7 +8,8 @@ Who Need Help is a modular Phoenix application rather than a collection of
 premature microservices. It produces one immutable image with three runtime
 roles:
 
-- `web`: Phoenix Endpoint, LiveView, PubSub, and Presence.
+- `web`: Phoenix Endpoint, LiveView, PubSub, Presence, and a producerless Oban
+  client for atomic job insertion.
 - `worker`: Oban queues and scheduled jobs; no public HTTP listener.
 - `migrate`: a one-shot database migration command before rollout.
 
@@ -61,6 +62,9 @@ are separate operational work and are not represented as complete.
   and derived proximity signals.
 - `Trust`: reviews, reports, blocks, leaderboard/reputation projections,
   abuse signals, moderator audit events, and shared rate-limit policies.
+- `Push`: privacy-safe product event construction, unique durable Oban jobs,
+  and a provider-neutral delivery adapter. Current events cover request
+  acceptance and new matched-chat messages.
 
 Contexts normally call each other through public functions. A small number of
 documented trust-and-safety transactions update related schemas together when
@@ -80,8 +84,11 @@ No sticky session is required for authenticated requests. Session cookies are
 signed by a shared secret. Uploads, if later introduced, must go to shared
 object storage rather than a container filesystem.
 
-Oban is enabled only for the worker role. PostgreSQL coordinates queues and
+Web and migrate application processes start a producerless Oban client with no
-leadership, so no Redis dependency is introduced.
+queues, plugins, or peer leadership so transactions can insert unique jobs.
+Only the worker role starts queue consumers and scheduled-job plugins.
+PostgreSQL coordinates queues and leadership, so no Redis dependency is
+introduced.
 
 ## Geospatial data
 

docs/local-hardening-plan.md

@@ -10,12 +10,12 @@ item below unless the evidence column explicitly describes a local mock.
 | Browser E2E | Manual headed-Chrome scenarios exist; no committed browser suite | A fresh uniquely named Compose project runs two-user urgent help, Activity, moderation, privacy, and error paths; traces are retained on failure; its exact volume is removed |
 | Android UI | Two JVM unit-test files; no `androidTest` source set | Emulator instrumentation covers deep links, permissions, foreground tracking, notification Stop, lifecycle, and network failure |
 | CI and quality | No tracked CI workflow or static/security analysis dependencies | The same containerized gates pass locally and are represented in a validated CI workflow |
-| Localization and accessibility | Completed locally: product copy and custom validation messages are extracted; EN/UK/RU catalogs and localized category descriptions/structured values are implemented | 508 default and 40 error messages are current; RU/UK have no empty/fuzzy entries; 167 backend tests and all 8 browser specs pass, including locale persistence, keyboard, axe, themes, responsive widths, and reconnect |
+| Localization and accessibility | Completed locally: product copy and custom validation messages are extracted; EN/UK/RU catalogs and localized category descriptions/structured values are implemented | 508 default and 40 error messages are current; RU/UK have no empty/fuzzy entries; 171 backend tests and all 8 browser specs pass, including locale persistence, keyboard, axe, themes, responsive widths, and reconnect |
 | Database scale | Core discovery/chat/moderation lists call unbounded `Repo.all()` | Cursor-bounded queries pass behavior tests and measured `EXPLAIN ANALYZE` checks on an isolated generated dataset |
 | Load and resilience | Public/readiness/heartbeat k6 profile exists | Authenticated writes, chat, tracking, reconnect, rolling replacement, and worker retry profiles pass without touching staging data |
 | Observability | Completed locally: protected per-process metrics feed a pinned Prometheus/Grafana/Alertmanager profile | All 3 direct web targets are up before/after the drill; a verified replica stop delivers firing and resolved webhooks; Grafana datasource/dashboard and an empty DB-count diff are retained |
 | Backup | Completed locally: a plaintext-free Restic stream is retained in pinned local MinIO | The encrypted repository passes full-data checking and fresh-database restore; corrupted configuration and an interrupted upload fail closed, leave no snapshot, and their temporary buckets are removed |
-| External boundaries | Completed locally: real Assent/Req, Swoosh/gen_smtp, and the applicable future push adapter run against an internal protocol mock | Success, rejection, retry, replay, timeout, PKCE/state, one-time-code, and idempotency paths pass without host ports or retained secrets; remote push product integration is explicitly not implemented |
+| External boundaries | Completed locally: real Assent/Req, Swoosh/gen_smtp, and the provider-neutral push adapter run against an internal protocol mock | Success, rejection, retry, replay, timeout, PKCE/state, one-time-code, and idempotency paths pass without host ports or retained secrets; acceptance/chat product events pass through an ephemeral PostGIS database and two Oban workers |
 | Final regression | 167 Phoenix tests plus reproducible browser and Android device suites | Browser, Android, API, DB, WebSocket, backup, monitoring, failure, cleanup, docs, and clean Git are verified from the final commits |
 
 The goal remains open while any row lacks reproducible local evidence.
@@ -107,13 +107,16 @@ The goal remains open while any row lacks reproducible local evidence.
   temporary restore database, and left source table counts unchanged.
 - The isolated external-service drill uses the release's configured Assent/Req
   and Swoosh/gen_smtp clients against an internal-only protocol mock. Its
-  canonical run passed OAuth success, PKCE/state rejection, provider denial,
+  latest run passed OAuth success, PKCE/state rejection, provider denial,
   one-time-code replay, temporary failure and timeout; SMTP success, permanent
-  rejection, one temporary retry, timeout and repeated submission; and the
+  rejection, one temporary retry, timeout and repeated submission; and the HTTP
-  future HTTP push boundary's disabled default, retry and idempotent replay and
+  push boundary's disabled default, retry, rejection, replay, and ambiguous
-  timeout paths. It retained no credentials, published no host ports, removed
+  timeout paths. An ephemeral PostGIS database plus two worker replicas also
-  its exact project and one-run images, and records remote push workflow
+  delivered real acceptance and chat events, observed an Oban retry on attempt
-  integration as not implemented.
+  2, rejected replay before transport, and excluded private chat content. It
+  retained no credentials, published no host ports, and removed its exact
+  project, volume, network, and one-run images. FCM/APNs device registration is
+  not claimed.
 - The final regression repeated the 167-test quality/security gate, isolated
   browser 1/1 bootstrap plus 8/8 Chromium specs, Android debug/staging builds
   and 5/5 API 37 device tests, the 50,000-row database benchmark, authenticated

docs/operations.md

@@ -84,7 +84,7 @@ RTO, retention, capacity, key custody, object locking, or database HA.
 
 ## Local external-service boundary drill
 
-Run the OAuth, SMTP, and future push-adapter protocol checks without public
+Run the OAuth, SMTP, and provider-neutral push protocol checks without public
 credentials, a server, or host-published ports:
 
 ```bash
@@ -104,22 +104,28 @@ non-root standard-library Python protocol mock, and then verifies:
    submitting the same message twice;
 3. the disabled default push boundary plus HTTP success, permanent rejection,
    one temporary retry, replay deduplication, and deduplication after an
-   ambiguous timeout using the same idempotency key.
+   ambiguous timeout using the same idempotency key;
+4. a fresh PostGIS database, current migrations, and two Oban worker replicas;
+   request acceptance and new-chat domain transactions enqueue stable
+   user-recipient events, replay is deduplicated before HTTP, an injected
+   temporary chat delivery fails its first Oban attempt and completes on its
+   second, and private message text is absent from the push payload.
 
-The evidence JSON and mock state contain only counters, booleans, normalized
+The evidence JSON and mock state contain counters, booleans, normalized
-identity fields, and payload digests. The script fails if any generated secret
+identity fields, payload digests, run-scoped user/event identifiers, and the
-appears in retained evidence. Its trap validates exact Compose labels, removes
+privacy-safe notification metadata asserted by the drill. The script fails if
-only that project, volumes and one-run images, and deletes the temporary
+any generated secret appears in retained evidence. Its trap validates exact
-credential file. Failure logs are retained under the same ignored evidence
+Compose labels, removes only that project, volumes and one-run images, and
-directory.
+deletes the temporary credential file. Failure logs are retained under the
+same ignored evidence directory.
 
 This drill uses the application's real Assent/Req and Swoosh/gen_smtp clients,
-but the provider is local. It therefore verifies the client-side protocol
+but the providers are local. It therefore verifies client-side protocol and
-boundary, not GitHub availability or SMTP-provider reliability. SMTP permits
+product integration, not GitHub, SMTP-provider, FCM, or APNs availability.
-duplicate delivery after ambiguous outcomes, so the result explicitly makes no
+SMTP permits duplicate delivery after ambiguous outcomes, so the result
-exactly-once claim. The push HTTP adapter is not connected to request, chat, or
+explicitly makes no exactly-once claim. Push currently targets a stable
-tracking workflows and is not an FCM or APNs implementation; remote push
+`user:<uuid>` recipient; selecting a provider, registering device tokens, and
-product integration remains unimplemented.
+resolving that user to devices remain deployment/provider work.
 
 The implementation follows the configured adapter interfaces and protocol
 semantics documented by

docs/verification.md

@@ -16,17 +16,17 @@ results from product limits and unknown production properties.
 | Consent-driven live tracking | Implemented and cross-client verified | On API 37, Android started `TrackingService` as a location foreground service with a persistent Stop notification. After Home minimized the Activity, an emulator coordinate change reached PostGIS. Notification Stop removed the service, notification, active session, and raw position. | Browsers stop with the page. Android has no `ACCESS_BACKGROUND_LOCATION`, unattended start, or route history. |
 | Privacy settings | Implemented and browser-verified | The profile exposed hidden, approximate public, exact for active match, and explicit exact-public options. Blocking and current-position cleanup have automated tests. | Exact public location remains a user opt-in; legal privacy and retention text still requires jurisdiction-specific review before launch. |
 | Reputation and anti-abuse | Implemented at MVP level | Handover codes, two-party completion, double-blind reviews, unique-counterpart ranking, optional movement/proximity evidence, reports, blocks, abuse signals, and moderator audit paths have automated tests. | The system is not bot-proof and does not claim identity verification. No punitive numeric policy is enabled without measured and approved thresholds. |
-| Social profiles | Manual links implemented; GitHub verification implemented and automated-tested | Manual links cannot set verification fields. The optional GitHub flow uses state, PKCE, a user-bound one-time session, unique provider ownership, and an audit record; 167 tests pass, including callback replay/state checks. The local protocol drill also performs real HTTP token/user exchanges without returning an access token to the application. | The staging operator has not supplied GitHub OAuth credentials, so the real external provider redirect/callback remains disabled and has not been browser-verified. Other providers remain manual/unverified. |
+| Social profiles | Manual links implemented; GitHub verification implemented and automated-tested | Manual links cannot set verification fields. The optional GitHub flow uses state, PKCE, a user-bound one-time session, unique provider ownership, and an audit record; 171 tests pass, including callback replay/state checks. The local protocol drill also performs real HTTP token/user exchanges without returning an access token to the application. | The staging operator has not supplied GitHub OAuth credentials, so the real external provider redirect/callback remains disabled and has not been browser-verified. Other providers remain manual/unverified. |
 | Voluntary thanks | Implemented as an external optional link | A helper can expose an optional link after completion; the UI states that the platform does not process the payment. | The platform does not provide payments, escrow, refunds, tax reporting, or payment guarantees. |
 | Android client | Local and public-staging clients implemented and emulator-verified | The native packages `org.whoneedhelp.mobile.debug` and `org.whoneedhelp.mobile.staging` launch the same authenticated LiveView app. Public HTTPS login, map, two-way chat, permission prompts, minimized foreground-service location updates, notification Stop, deep-link routing, and server cleanup were exercised on API 37. | Production signing, Play Store publication, verified Android App Links, unattended/background-permission tracking, and iOS are not implemented. |
 | Multiple web/worker instances | Implemented and locally failure/rollout-verified | The isolated Compose profile passed BEAM crashes and sequential replacement with 3 web/2 worker replicas, all five nodes joined, PubSub passed, and 758/758 readiness requests succeeded. The project-owned kind cluster replaced all 2 web/2 worker pod UIDs under `maxUnavailable=0`; all four replacement pods joined and PubSub passed. | Local PostGIS is a single instance. Production database HA, backups, and recovery are operator work and are not claimed complete. |
 | Local observability | Implemented and protocol-verified | Pinned Prometheus scraped all 3 direct load web targets with a file Bearer credential; Grafana provisioned a healthy datasource and four-panel dashboard; Alertmanager delivered firing and resolved webhooks for an induced scoped replica stop. | Local delivery does not establish production retention, notification-provider reliability, on-call policy, or measured alert thresholds. |
 | Encrypted local backup | Implemented and failure-verified | Pinned Restic streamed PostgreSQL custom format into pinned local MinIO with no host plaintext dump, passed full-data checking and a fresh-database restore, rejected a corrupted repository, and published no snapshot for an interrupted upload. | The retained MinIO volume is on the same workstation; this is not off-site storage, database HA, or a production RPO/RTO/retention claim. |
-| External protocol boundaries | Implemented and locally failure-verified | The production release used its configured Assent/Req and Swoosh/gen_smtp clients against an internal-only mock; OAuth and SMTP success/rejection/retry/replay/timeout paths passed. The future HTTP push boundary passed disabled, retry, rejection and idempotency paths. | This does not verify external provider availability. Remote FCM/APNs delivery and product-workflow integration are not implemented; SMTP exactly-once delivery is not claimed. |
+| External protocol boundaries | Implemented and locally failure-verified | The production release used its configured Assent/Req and Swoosh/gen_smtp clients against an internal-only mock; OAuth and SMTP success/rejection/retry/replay/timeout paths passed. The HTTP push boundary passed disabled, retry, rejection, timeout, and idempotency paths. Request acceptance and new-chat transactions created durable jobs processed by two Oban worker replicas; the chat event completed on Oban attempt 2 after an injected temporary failure. | This does not verify external provider availability or device delivery. FCM/APNs token registration and provider selection remain external work; SMTP exactly-once delivery is not claimed. |
 
 ## Reproducible checks
 
-- The isolated Phoenix suite completed on 2026-07-19 with 167
+- The isolated Phoenix suite completed on 2026-07-19 with 171
   tests and 0 failures after cursor pagination, database aggregation, and the
   full localization changes
   on Elixir 1.20.2 and Erlang/OTP 29.0.3.
@@ -114,17 +114,20 @@ results from product limits and unknown production properties.
   canonical bucket remains. Evidence is retained at
   `output/backups-s3/final-backup-20260719a`; the retained MinIO bucket is
   `wnh-backup-final-backup-20260719a`.
-- The canonical external-boundary drill passed real client-side OAuth token and
+- The latest external-boundary drill passed real client-side OAuth token and
   user HTTP exchanges, PKCE/state checks, provider denial, one-time-code replay,
   a fresh flow after a temporary token error, and timeout failure. It also
   passed SMTP acceptance, permanent rejection without retry, one retry after a
-  temporary greeting, timeout, and repeated submission, plus the future push
+  temporary greeting, timeout, and repeated submission. The push checks passed
-  adapter's disabled default, rejection, one retry, replay deduplication and
+  the disabled adapter, rejection, HTTP retry, replay and ambiguous-timeout
-  ambiguous-timeout deduplication. Every retained file is mode `0600`, the
+  deduplication, then created real acceptance/chat events against an ephemeral
-  evidence contains no generated credential, no host port was published, and
+  PostGIS database. Two worker replicas processed them; acceptance completed on
-  the exact Compose project, network, and one-run images were absent after
+  attempt 1, chat completed on Oban attempt 2 after a temporary provider
-  cleanup. Evidence is retained at
+  failure, replay never reached HTTP, and the chat text was absent from the
-  `output/external-boundaries/boundary-canonical-20260719d`.
+  payload. Every retained file is mode `0600`, no generated credential was
+  retained, no host port was published, and the exact project, volume, network,
+  and one-run images were absent after cleanup. Evidence is retained at
+  `output/external-boundaries/push-product-pass3-20260719`.
 - The committed browser suite passed its 1/1 bootstrap and all 8/8 Chromium
   specs against a fresh PostGIS volume with two web and two worker replicas on
   2026-07-19. The retained successful-run artifact directory is

lib/who_need_help/application.ex

@@ -14,11 +14,14 @@ defmodule WhoNeedHelp.Application do
       {Phoenix.PubSub, name: WhoNeedHelp.PubSub}
     ]
 
+    oban_config = Application.fetch_env!(:who_need_help, Oban)
+    oban_client_config = Keyword.merge(oban_config, queues: [], plugins: [], peer: false)
+
     role_children =
       case Application.fetch_env!(:who_need_help, :app_role) do
-        :web -> [WhoNeedHelpWeb.Presence, WhoNeedHelpWeb.Endpoint]
+        :web -> [{Oban, oban_client_config}, WhoNeedHelpWeb.Presence, WhoNeedHelpWeb.Endpoint]
-        :worker -> [{Oban, Application.fetch_env!(:who_need_help, Oban)}]
+        :worker -> [{Oban, oban_config}]
-        :migrate -> []
+        :migrate -> [{Oban, oban_client_config}]
       end
 
     children = common_children ++ role_children

lib/who_need_help/external_boundary_drill.ex

@@ -1,15 +1,22 @@
 defmodule WhoNeedHelp.ExternalBoundaryDrill do
   @moduledoc """
-  Explicit local-only protocol drill for OAuth, SMTP, and the future push boundary.
+  Explicit local-only protocol drill for OAuth, SMTP, and the push boundary.
 
   It is invoked by `scripts/external-boundaries-run.sh`; normal application
   startup never calls it.
   """
 
+  import Ecto.Query, only: [where: 3]
   import Swoosh.Email
 
+  alias WhoNeedHelp.Accounts
+  alias WhoNeedHelp.Accounts.{Scope, User}
+  alias WhoNeedHelp.{Catalog, Help, Messaging}
   alias WhoNeedHelp.Push
+  alias WhoNeedHelp.Push.DeliveryWorker
+  alias WhoNeedHelp.Push.DisabledAdapter, as: PushDisabledAdapter
   alias WhoNeedHelp.Push.HTTPAdapter, as: PushHTTPAdapter
+  alias WhoNeedHelp.Repo
   alias WhoNeedHelp.SocialOAuth.AssentAdapter
 
   @oauth_redirect_uri "http://boundary.local/auth/social/github/callback"
@@ -20,15 +27,18 @@ defmodule WhoNeedHelp.ExternalBoundaryDrill do
     ensure_started!(:req)
     ensure_started!(:swoosh)
     ensure_started!(:gen_smtp)
+    ensure_started!(:who_need_help)
 
     base_url = fetch_env!("EXTERNAL_MOCK_BASE_URL")
+    protocol_push = push_drill(base_url)
+    product_push = push_product_drill(base_url)
 
     summary = %{
       status: "passed",
       oauth: oauth_drill(base_url),
       smtp: smtp_drill(base_url),
-      push: push_drill(base_url),
+      push: Map.put(protocol_push, :domain_workflow_integration, product_push.status),
-      push_product_integration: "not_implemented"
+      push_product_integration: product_push
     }
 
     File.mkdir_p!(@output_directory)
@@ -286,7 +296,7 @@ defmodule WhoNeedHelp.ExternalBoundaryDrill do
     bearer_token = fetch_env!("EXTERNAL_PUSH_BEARER_TOKEN")
 
     assert_error!(
-      Push.deliver(push_notification("disabled")),
+      Push.deliver(push_notification("disabled"), adapter: PushDisabledAdapter),
       "default push adapter unexpectedly delivered"
     )
 
@@ -351,11 +361,182 @@ defmodule WhoNeedHelp.ExternalBoundaryDrill do
       permanent_rejection_not_retried: true,
       temporary_failure_retried_once: true,
       replay_deduplicated: true,
-      timeout_after_accept_deduplicated: true,
+      timeout_after_accept_deduplicated: true
-      domain_workflow_integration: "not_implemented"
     }
   end
 
+  defp push_product_drill(base_url) do
+    expected_worker_replicas =
+      fetch_env!("EXTERNAL_EXPECTED_WORKER_REPLICAS")
+      |> parse_positive_integer!("EXTERNAL_EXPECTED_WORKER_REPLICAS")
+
+    requester = confirmed_user!("requester")
+    helper = confirmed_user!("helper")
+    requester_scope = Scope.for_user(requester)
+    helper_scope = Scope.for_user(helper)
+    category = Catalog.seed_defaults()
+
+    control!(base_url, "push", "success")
+
+    {:ok, request} =
+      Help.create_request(requester_scope, %{
+        "title" => "Boundary medicine pickup",
+        "description" => "The reserved legal medicine is ready for pickup.",
+        "pickup_instructions" => "Ask for the run-scoped reservation.",
+        "location_label" => "Boundary district",
+        "latitude" => "50.4501",
+        "longitude" => "30.5234",
+        "urgency" => "now",
+        "location_visibility" => "approximate_public",
+        "structured_data" => %{"pickup_status" => "reserved"},
+        "expires_at" => DateTime.utc_now(:second) |> DateTime.add(3, :hour),
+        "category_id" => category.id
+      })
+
+    {:ok, assignment} = Help.accept_request(helper_scope, request.id)
+
+    acceptance_key = "request-accepted:#{assignment.id}:#{requester.id}"
+    acceptance_job = wait_for_completed_job!(acceptance_key, 1)
+    acceptance_state = state!(base_url)["push"]
+
+    assert!(acceptance_job.attempt == 1, "acceptance push was not completed on its first job run")
+    assert!(acceptance_state["attempts"] == 1, "acceptance push attempt count was not one")
+    assert!(acceptance_state["deliveries"] == 1, "acceptance push delivery count was not one")
+
+    [acceptance_notification] = acceptance_state["notifications"]
+
+    assert!(
+      acceptance_notification["data"]["kind"] == "request_accepted",
+      "acceptance product event kind was not delivered"
+    )
+
+    assert!(
+      acceptance_notification["recipient"] == "user:#{requester.id}",
+      "acceptance product event targeted the wrong user"
+    )
+
+    {:ok, replay_job} =
+      Push.enqueue_request_accepted(assignment.id, request.id, requester.id)
+
+    assert!(replay_job.conflict?, "product event replay created a new Oban job")
+    assert!(replay_job.id == acceptance_job.id, "product event replay returned another job")
+    Process.sleep(250)
+
+    acceptance_replay_state = state!(base_url)["push"]
+
+    assert!(
+      acceptance_replay_state["attempts"] == 1,
+      "product event replay reached the transport boundary"
+    )
+
+    control!(base_url, "push", "temporary_once")
+    message_body = "Run-scoped chat text must not enter push"
+
+    {:ok, message} =
+      Messaging.send_message(requester_scope, assignment, %{"body" => message_body})
+
+    message_key = "message-created:#{message.id}:#{helper.id}"
+    message_job = wait_for_completed_job!(message_key, 2)
+    message_state = state!(base_url)["push"]
+
+    assert!(message_job.attempt == 2, "chat push did not complete on the second Oban attempt")
+    assert!(message_state["attempts"] == 2, "chat push temporary failure was not retried")
+    assert!(message_state["deliveries"] == 1, "chat push retry duplicated delivery")
+
+    [message_notification] = message_state["notifications"]
+
+    assert!(
+      message_notification["data"]["kind"] == "message_created",
+      "chat product event kind was not delivered"
+    )
+
+    assert!(
+      message_notification["recipient"] == "user:#{helper.id}",
+      "chat product event targeted the wrong user"
+    )
+
+    assert!(
+      message_notification["body"] != message_body and
+        not String.contains?(Jason.encode!(message_notification), message_body),
+      "chat message content escaped into the push payload"
+    )
+
+    %{
+      status: "passed",
+      acceptance_delivery: "passed",
+      chat_delivery: "passed",
+      oban_retry_observed: true,
+      replay_deduplicated_before_transport: true,
+      message_content_excluded: true,
+      expected_worker_replicas: expected_worker_replicas,
+      database_scope: "isolated_ephemeral_volume"
+    }
+  end
+
+  defp confirmed_user!(label) do
+    suffix = Ecto.UUID.generate()
+
+    {:ok, user} =
+      Accounts.register_user(%{
+        email: "#{label}-#{suffix}@boundary.invalid",
+        display_name: "Boundary #{label}",
+        terms_accepted: true
+      })
+
+    user
+    |> User.confirm_changeset()
+    |> Repo.update!()
+  end
+
+  defp wait_for_completed_job!(idempotency_key, minimum_attempts) do
+    deadline = System.monotonic_time(:millisecond) + 60_000
+    do_wait_for_completed_job!(idempotency_key, minimum_attempts, deadline)
+  end
+
+  defp do_wait_for_completed_job!(idempotency_key, minimum_attempts, deadline) do
+    worker_name = DeliveryWorker.__opts__() |> Keyword.fetch!(:worker)
+
+    job =
+      Oban.Job
+      |> where(
+        [job],
+        job.worker == ^worker_name and
+          fragment("?->>'idempotency_key' = ?", job.args, ^idempotency_key)
+      )
+      |> Repo.one()
+
+    cond do
+      job && job.state == "completed" && job.attempt >= minimum_attempts ->
+        job
+
+      job && job.state in ["cancelled", "discarded"] ->
+        raise "push product job ended in #{job.state}"
+
+      System.monotonic_time(:millisecond) >= deadline ->
+        observed =
+          case job do
+            nil ->
+              "missing"
+
+            job ->
+              inspect(%{
+                id: job.id,
+                state: job.state,
+                attempt: job.attempt,
+                max_attempts: job.max_attempts,
+                queue: job.queue,
+                scheduled_at: job.scheduled_at
+              })
+          end
+
+        raise "push product job did not complete within the local drill window: #{observed}"
+
+      true ->
+        Process.sleep(100)
+        do_wait_for_completed_job!(idempotency_key, minimum_attempts, deadline)
+    end
+  end
+
   defp push_notification(idempotency_key) do
     %{
       idempotency_key: idempotency_key,
@@ -424,6 +605,13 @@ defmodule WhoNeedHelp.ExternalBoundaryDrill do
     end
   end
 
+  defp parse_positive_integer!(value, name) do
+    case Integer.parse(value) do
+      {integer, ""} when integer > 0 -> integer
+      _other -> raise "#{name} must be a positive integer."
+    end
+  end
+
   defp ensure_started!(application) do
     case Application.ensure_all_started(application) do
       {:ok, _started} -> :ok

lib/who_need_help/help.ex

@@ -9,6 +9,7 @@ defmodule WhoNeedHelp.Help do
   alias WhoNeedHelp.Catalog.Category
   alias WhoNeedHelp.Help.{Assignment, HelpRequest}
   alias WhoNeedHelp.Pagination
+  alias WhoNeedHelp.Push
   alias WhoNeedHelp.Repo
   alias WhoNeedHelp.Trust
   alias WhoNeedHelp.Trust.Block
@@ -230,7 +231,13 @@ defmodule WhoNeedHelp.Help do
                    {:ok, _audit} <-
                      Trust.audit(helper.id, "request.accepted", "assignment", assignment.id, %{
                        "request_id" => request.id
-                     }) do
+                     }),
+                   {:ok, _push_job} <-
+                     Push.enqueue_request_accepted(
+                       assignment.id,
+                       request.id,
+                       request.requester_id
+                     ) do
                 {:ok, assignment}
               end
           end

lib/who_need_help/messaging.ex

@@ -7,6 +7,7 @@ defmodule WhoNeedHelp.Messaging do
   alias WhoNeedHelp.Help.Assignment
   alias WhoNeedHelp.Messaging.Message
   alias WhoNeedHelp.Pagination
+  alias WhoNeedHelp.Push
   alias WhoNeedHelp.Repo
   alias WhoNeedHelp.Trust
 
@@ -44,10 +45,25 @@ defmodule WhoNeedHelp.Messaging do
     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)
+      recipient_id = counterpart_id(user.id, assignment, request)
+
       result =
+        Repo.transact(fn ->
+          with {:ok, message} <-
                  %Message{assignment_id: assignment.id, sender_id: user.id}
                  |> Message.changeset(attrs)
-        |> Repo.insert()
+                 |> Repo.insert(),
+               {:ok, _push_job} <-
+                 Push.enqueue_message_created(
+                   message.id,
+                   assignment.id,
+                   request.id,
+                   recipient_id
+                 ) do
+            {:ok, message}
+          end
+        end)
 
       with {:ok, message} <- result do
         message = Repo.preload(message, :sender)
@@ -69,11 +85,11 @@ defmodule WhoNeedHelp.Messaging do
 
   defp blocked_assignment?(%Scope{user: user}, assignment) do
     request = Repo.preload(assignment, :request).request
+    Trust.blocked_between?(user.id, counterpart_id(user.id, assignment, request))
+  end
 
-    counterpart_id =
+  defp counterpart_id(user_id, assignment, request) do
-      if user.id == assignment.helper_id, do: request.requester_id, else: assignment.helper_id
+    if user_id == assignment.helper_id, do: request.requester_id, else: assignment.helper_id
-
-    Trust.blocked_between?(user.id, counterpart_id)
   end
 
   defp before_message(query, nil), do: query

lib/who_need_help/push.ex

@@ -1,13 +1,15 @@
 defmodule WhoNeedHelp.Push do
   @moduledoc """
-  Provider-neutral boundary for future remote push delivery.
+  Provider-neutral boundary and durable product-event enqueueing for remote push.
 
-  The product does not currently register FCM/APNs device tokens or invoke this
+  The product deliberately remains disabled until a complete HTTP adapter
-  boundary from domain workflows. The default adapter is deliberately disabled.
+  configuration is supplied. Product events target a stable user identifier;
-  A local HTTP adapter exists so delivery, retry, timeout, and idempotency
+  the selected provider is responsible for resolving that identifier to one or
-  semantics can be verified before a provider is selected.
+  more registered devices.
   """
 
+  alias WhoNeedHelp.Push.DeliveryWorker
+
   @type notification :: %{
           required(:idempotency_key) => String.t(),
           required(:recipient) => String.t(),
@@ -22,6 +24,47 @@ defmodule WhoNeedHelp.Push do
   @callback deliver(notification(), keyword()) ::
               {:ok, receipt()} | {:error, delivery_error()}
 
+  def enabled?, do: Application.get_env(:who_need_help, :push_product_enabled, false) == true
+
+  def enqueue_request_accepted(assignment_id, request_id, requester_id) do
+    enqueue(%{
+      idempotency_key: "request-accepted:#{assignment_id}:#{requester_id}",
+      recipient: user_recipient(requester_id),
+      title: "A helper responded",
+      body: "Open Who Need Help to see the request update.",
+      data: %{
+        "kind" => "request_accepted",
+        "assignment_id" => assignment_id,
+        "request_id" => request_id
+      }
+    })
+  end
+
+  def enqueue_message_created(message_id, assignment_id, request_id, recipient_id) do
+    enqueue(%{
+      idempotency_key: "message-created:#{message_id}:#{recipient_id}",
+      recipient: user_recipient(recipient_id),
+      title: "New message",
+      body: "Open Who Need Help to read the conversation.",
+      data: %{
+        "kind" => "message_created",
+        "assignment_id" => assignment_id,
+        "message_id" => message_id,
+        "request_id" => request_id
+      }
+    })
+  end
+
+  def enqueue(notification) do
+    if enabled?() do
+      notification
+      |> DeliveryWorker.new()
+      |> Oban.insert()
+    else
+      {:ok, :disabled}
+    end
+  end
+
   def deliver(notification, opts \\ []) do
     adapter =
       Keyword.get_lazy(opts, :adapter, fn ->
@@ -37,6 +80,12 @@ defmodule WhoNeedHelp.Push do
     end
   end
 
+  def delivery_options do
+    Application.get_env(:who_need_help, :push_delivery_options, [])
+  end
+
+  defp user_recipient(user_id), do: "user:#{user_id}"
+
   defp validate(%{
          idempotency_key: idempotency_key,
          recipient: recipient,

lib/who_need_help/push/delivery_worker.ex

@@ -0,0 +1,58 @@
+defmodule WhoNeedHelp.Push.DeliveryWorker do
+  @moduledoc """
+  Delivers product push events from PostgreSQL through the configured boundary.
+
+  Event keys remain unique for the lifetime of retained Oban rows. Permanent
+  configuration/provider rejections are cancelled; transport and retryable HTTP
+  failures use Oban's retry policy.
+  """
+
+  use Oban.Worker,
+    queue: :push,
+    unique: [
+      period: :infinity,
+      fields: [:args, :worker],
+      keys: [:idempotency_key]
+    ]
+
+  alias WhoNeedHelp.Push
+
+  @impl Oban.Worker
+  def perform(%Oban.Job{
+        args:
+          %{
+            "idempotency_key" => idempotency_key,
+            "recipient" => recipient,
+            "title" => title,
+            "body" => body
+          } = args
+      }) do
+    notification = %{
+      idempotency_key: idempotency_key,
+      recipient: recipient,
+      title: title,
+      body: body,
+      data: Map.get(args, "data", %{})
+    }
+
+    case Push.deliver(notification, Push.delivery_options()) do
+      {:ok, _receipt} ->
+        :ok
+
+      {:error, :disabled} ->
+        {:cancel, :push_disabled}
+
+      {:error, {:invalid_notification, _field} = reason} ->
+        {:cancel, reason}
+
+      {:error, {:invalid_option, _option} = reason} ->
+        {:cancel, reason}
+
+      {:error, {:rejected, _status, _body} = reason} ->
+        {:cancel, reason}
+
+      {:error, reason} ->
+        {:error, reason}
+    end
+  end
+end

ops/external-boundaries/mock_server.py

@@ -37,6 +37,7 @@ class BoundaryState:
         self.push_attempts = 0
         self.push_receipts: dict[str, str] = {}
         self.push_deliveries = 0
+        self.push_notifications: list[dict[str, Any]] = []
         self.smtp_connections = {
             str(SMTP_PORT): 0,
             str(SMTP_RETRY_PORT): 0,
@@ -60,6 +61,7 @@ class BoundaryState:
                 self.push_attempts = 0
                 self.push_receipts = {}
                 self.push_deliveries = 0
+                self.push_notifications = []
             elif component == "smtp":
                 self.smtp_connections = {
                     str(SMTP_PORT): 0,
@@ -90,6 +92,7 @@ class BoundaryState:
                     "attempts": self.push_attempts,
                     "deliveries": self.push_deliveries,
                     "idempotency_keys": len(self.push_receipts),
+                    "notifications": list(self.push_notifications),
                 },
                 "smtp": {
                     "connections": dict(self.smtp_connections),
@@ -355,6 +358,15 @@ class BoundaryHTTPHandler(BaseHTTPRequestHandler):
                 receipt = f"push-receipt-{len(STATE.push_receipts) + 1}"
                 STATE.push_receipts[idempotency_key] = receipt
                 STATE.push_deliveries += 1
+                STATE.push_notifications.append(
+                    {
+                        "idempotency_key": idempotency_key,
+                        "recipient": notification["recipient"],
+                        "title": notification["title"],
+                        "body": notification["body"],
+                        "data": notification.get("data", {}),
+                    }
+                )
 
         if mode == "timeout_after_accept" and attempt == 1:
             time.sleep(DELAY_SECONDS)

scripts/external-boundaries-run.sh

@@ -51,7 +51,8 @@ external_boundary_mock_image="who-need-help:boundary-mock-$run_id"
   printf 'EXTERNAL_OAUTH_CLIENT_ID=%s\n' "$oauth_client_id"
   printf 'EXTERNAL_OAUTH_CLIENT_SECRET=%s\n' "$oauth_client_secret"
   printf 'EXTERNAL_PUSH_BEARER_TOKEN=%s\n' "$push_bearer_token"
-  printf 'EXTERNAL_DATABASE_URL=ecto://boundary:%s@unused/unused\n' "$database_password"
+  printf 'EXTERNAL_POSTGRES_PASSWORD=%s\n' "$database_password"
+  printf 'EXTERNAL_DATABASE_URL=ecto://boundary:%s@boundary-db/boundary\n' "$database_password"
   printf 'EXTERNAL_SECRET_KEY_BASE=%s\n' "$secret_key_base"
   printf 'EXTERNAL_HANDOVER_SECRET=%s\n' "$handover_secret"
 } >"$env_file"
@@ -109,7 +110,23 @@ trap cleanup EXIT HUP INT TERM
 
 "${compose[@]}" config --quiet
 "${compose[@]}" build external-mock boundary-check >"$output_dir/build.log"
-"${compose[@]}" up --detach --wait external-mock >"$output_dir/mock-up.log"
+"${compose[@]}" up --detach --wait external-mock boundary-db boundary-migrate \
+  >"$output_dir/dependencies-up.log"
+"${compose[@]}" up --detach --wait --scale boundary-worker=2 boundary-worker \
+  >"$output_dir/workers-up.log"
+
+mapfile -t worker_ids < <("${compose[@]}" ps --quiet boundary-worker)
+
+if ((${#worker_ids[@]} != 2)); then
+  echo "Expected two running boundary worker replicas; observed ${#worker_ids[@]}." >&2
+  exit 1
+fi
+
+for worker_id in "${worker_ids[@]}"; do
+  assert_project_scope "$worker_id"
+  test "$(docker inspect --format '{{.State.Running}}' "$worker_id")" = "true"
+done
+
 "${compose[@]}" run --rm --no-deps -T boundary-check \
   >"$output_dir/drill.log" 2>&1
 
@@ -139,8 +156,15 @@ jq -e '
   .push.temporary_failure_retried_once == true and
   .push.replay_deduplicated == true and
   .push.timeout_after_accept_deduplicated == true and
-  .push.domain_workflow_integration == "not_implemented" and
+  .push.domain_workflow_integration == "passed" and
-  .push_product_integration == "not_implemented"
+  .push_product_integration.status == "passed" and
+  .push_product_integration.acceptance_delivery == "passed" and
+  .push_product_integration.chat_delivery == "passed" and
+  .push_product_integration.oban_retry_observed == true and
+  .push_product_integration.replay_deduplicated_before_transport == true and
+  .push_product_integration.message_content_excluded == true and
+  .push_product_integration.expected_worker_replicas == 2 and
+  .push_product_integration.database_scope == "isolated_ephemeral_volume"
 ' "$output_dir/summary.json" >/dev/null
 
 for secret in "$oauth_client_secret" "$push_bearer_token" "$database_password" \

scripts/quality.sh

@@ -100,7 +100,8 @@ EXTERNAL_BOUNDARY_HOST_GID="$(id -g)" \
 EXTERNAL_OAUTH_CLIENT_ID=render-client \
 EXTERNAL_OAUTH_CLIENT_SECRET=render-secret \
 EXTERNAL_PUSH_BEARER_TOKEN=render-push-token \
-EXTERNAL_DATABASE_URL=ecto://render:render@unused/unused \
+EXTERNAL_POSTGRES_PASSWORD=render-database-secret \
+EXTERNAL_DATABASE_URL=ecto://boundary:render-database-secret@boundary-db/boundary \
 EXTERNAL_SECRET_KEY_BASE=render-secret-key-base \
 EXTERNAL_HANDOVER_SECRET=render-handover-secret \
   docker compose -f compose.external-boundaries.yaml config --quiet

test/who_need_help/push_product_test.exs

@@ -0,0 +1,155 @@
+defmodule WhoNeedHelp.PushProductTest do
+  use WhoNeedHelp.DataCase, async: false
+  use Oban.Testing, repo: WhoNeedHelp.Repo
+
+  import WhoNeedHelp.AccountsFixtures
+
+  alias WhoNeedHelp.{Catalog, Help, Messaging, Push}
+  alias WhoNeedHelp.Push.DeliveryWorker
+
+  defmodule RecordingAdapter do
+    @behaviour Push
+
+    @impl true
+    def deliver(notification, opts) do
+      send(Keyword.fetch!(opts, :test_pid), {:push_delivered, notification})
+      {:ok, %{id: "local-receipt", duplicate: false}}
+    end
+  end
+
+  setup do
+    keys = [:push_product_enabled, :push_adapter, :push_delivery_options]
+    previous = Map.new(keys, &{&1, Application.fetch_env(:who_need_help, &1)})
+
+    on_exit(fn ->
+      Enum.each(previous, fn
+        {key, {:ok, value}} -> Application.put_env(:who_need_help, key, value)
+        {key, :error} -> Application.delete_env(:who_need_help, key)
+      end)
+    end)
+
+    category = Catalog.seed_defaults()
+    requester = user_fixture(display_name: "Push requester")
+    helper = user_fixture(display_name: "Push helper")
+
+    attrs = %{
+      "title" => "Reserved medicine",
+      "description" => "Pickup is ready.",
+      "pickup_instructions" => "Ask for the reserved order.",
+      "location_label" => "Central district",
+      "latitude" => "50.4501",
+      "longitude" => "30.5234",
+      "urgency" => "now",
+      "location_visibility" => "approximate_public",
+      "structured_data" => %{"pickup_status" => "reserved"},
+      "expires_at" => DateTime.utc_now(:second) |> DateTime.add(3, :hour),
+      "category_id" => category.id
+    }
+
+    {:ok,
+     requester: requester,
+     helper: helper,
+     requester_scope: user_scope_fixture(requester),
+     helper_scope: user_scope_fixture(helper),
+     request_attrs: attrs}
+  end
+
+  test "acceptance and chat atomically enqueue privacy-safe recipient events", context do
+    Application.put_env(:who_need_help, :push_product_enabled, true)
+
+    {:ok, request} = Help.create_request(context.requester_scope, context.request_attrs)
+    {:ok, assignment} = Help.accept_request(context.helper_scope, request.id)
+
+    assert_enqueued(
+      worker: DeliveryWorker,
+      queue: :push,
+      args: %{
+        "idempotency_key" => "request-accepted:#{assignment.id}:#{context.requester.id}",
+        "recipient" => "user:#{context.requester.id}",
+        "title" => "A helper responded",
+        "body" => "Open Who Need Help to see the request update.",
+        "data" => %{
+          "kind" => "request_accepted",
+          "assignment_id" => assignment.id,
+          "request_id" => request.id
+        }
+      }
+    )
+
+    {:ok, message} =
+      Messaging.send_message(context.requester_scope, assignment, %{"body" => "Thank you"})
+
+    assert_enqueued(
+      worker: DeliveryWorker,
+      queue: :push,
+      args: %{
+        "idempotency_key" => "message-created:#{message.id}:#{context.helper.id}",
+        "recipient" => "user:#{context.helper.id}",
+        "title" => "New message",
+        "body" => "Open Who Need Help to read the conversation.",
+        "data" => %{
+          "kind" => "message_created",
+          "assignment_id" => assignment.id,
+          "message_id" => message.id,
+          "request_id" => request.id
+        }
+      }
+    )
+
+    refute Enum.any?(all_enqueued(worker: DeliveryWorker), fn job ->
+             inspect(job.args) =~ "Thank you"
+           end)
+  end
+
+  test "event idempotency key creates only one retained Oban job", context do
+    Application.put_env(:who_need_help, :push_product_enabled, true)
+
+    key = "request-accepted:#{Ecto.UUID.generate()}:#{context.requester.id}"
+    [_, assignment_id, requester_id] = String.split(key, ":")
+    request_id = Ecto.UUID.generate()
+
+    assert {:ok, first} =
+             Push.enqueue_request_accepted(assignment_id, request_id, requester_id)
+
+    assert {:ok, replay} =
+             Push.enqueue_request_accepted(assignment_id, request_id, requester_id)
+
+    assert replay.conflict?
+    assert replay.id == first.id
+    assert 1 == length(all_enqueued(worker: DeliveryWorker, args: %{"idempotency_key" => key}))
+  end
+
+  test "disabled product boundary leaves domain events job-free", context do
+    Application.put_env(:who_need_help, :push_product_enabled, false)
+
+    {:ok, request} = Help.create_request(context.requester_scope, context.request_attrs)
+    {:ok, assignment} = Help.accept_request(context.helper_scope, request.id)
+
+    {:ok, _message} =
+      Messaging.send_message(context.helper_scope, assignment, %{"body" => "On my way"})
+
+    assert [] == all_enqueued(worker: DeliveryWorker)
+  end
+
+  test "delivery worker reconstructs and sends the persisted notification" do
+    Application.put_env(:who_need_help, :push_adapter, RecordingAdapter)
+    Application.put_env(:who_need_help, :push_delivery_options, test_pid: self())
+
+    args = %{
+      "idempotency_key" => "message-created:event:user",
+      "recipient" => "user:recipient",
+      "title" => "New message",
+      "body" => "Open Who Need Help to read the conversation.",
+      "data" => %{"kind" => "message_created"}
+    }
+
+    assert :ok = perform_job(DeliveryWorker, args)
+
+    assert_received {:push_delivered,
+                     %{
+                       idempotency_key: "message-created:event:user",
+                       recipient: "user:recipient",
+                       data: %{"kind" => "message_created"}
+                     }}
+  end
+end