Stop sending push notifications to banned users (#338 )

Banning a user deletes their sessions and closes their connections but keeps their push subscriptions, and Room::MessagePusher chose recipients by membership alone. A banned user's browser or phone therefore went on receiving the room name, sender and text of new direct messages, mentions, and messages in rooms they had set to everything. Choose subscriptions from active users only. The subscriptions are kept, so unbanning brings notifications back without the user having to subscribe again (the client doesn't resubscribe while the browser still holds a subscription). Co-authored-by: Marcello Costagliola <176920116+namespaceMarcello@users.noreply.github.com>
Update benchmarks from the shared response-verified comparison
2026-10-28 00:16:48 +09:00 · 2026-10-07 11:25:16 -07:00 · 2026-10-07 17:53:35 +02:00 · 2026-10-07 17:08:48 +02:00 · 2026-10-07 16:55:08 +02:00 · 2026-10-07 16:37:53 +02:00
95 changed files with 2166 additions and 217 deletions
@@ -21,7 +21,7 @@ jobs:
          persist-credentials: false

      - name: Set up Ruby
-        uses: ruby/setup-ruby@a0102e0972be65f351c307e2d64b9314a57c8073 # v1.324.0
+        uses: ruby/setup-ruby@14594264cd68ce8a2345dd349bc3d138a4ef85c8 # v1.327.0
        with:
          ruby-version: .ruby-version
          bundler-cache: true
@@ -40,7 +40,7 @@ jobs:
          persist-credentials: false

      - name: Set up Ruby
-        uses: ruby/setup-ruby@a0102e0972be65f351c307e2d64b9314a57c8073 # v1.324.0
+        uses: ruby/setup-ruby@14594264cd68ce8a2345dd349bc3d138a4ef85c8 # v1.327.0
        with:
          ruby-version: .ruby-version
          bundler-cache: true
@@ -48,6 +48,29 @@ jobs:
      - name: Lint code for consistent style
        run: bin/rubocop

+  herb:
+    name: Herb check
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+    steps:
+      - name: Install packages
+        run: sudo apt-get update && sudo apt-get install --no-install-recommends -y libvips
+
+      - name: Checkout code
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+
+      - name: Set up Ruby
+        uses: ruby/setup-ruby@a0102e0972be65f351c307e2d64b9314a57c8073 # v1.324.0
+        with:
+          ruby-version: .ruby-version
+          bundler-cache: true
+
+      - name: Check that every HTML template compiles through Herb
+        run: bin/rails herb:check
+
  lint-actions:
    name: GitHub Actions audit
    runs-on: ubuntu-latest
@@ -79,7 +102,7 @@ jobs:
        options: --health-cmd "redis-cli ping" --health-interval 10s --health-timeout 5s --health-retries 5
    steps:
      - name: Install packages
-        run: sudo apt-get update && sudo apt-get install --no-install-recommends -y libsqlite3-0 libvips curl ffmpeg
+        run: sudo apt-get update && sudo apt-get install --no-install-recommends -y libsqlite3-0 sqlite3 libvips curl ffmpeg

      - name: Checkout code
        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
@@ -87,7 +110,7 @@ jobs:
          persist-credentials: false

      - name: Set up Ruby
-        uses: ruby/setup-ruby@a0102e0972be65f351c307e2d64b9314a57c8073 # v1.324.0
+        uses: ruby/setup-ruby@14594264cd68ce8a2345dd349bc3d138a4ef85c8 # v1.327.0
        env:
          REDIS_URL: redis://localhost:6379/0
        with:
@@ -97,6 +120,9 @@ jobs:
      - name: Run tests
        run: bin/rails db:setup test

+      - name: Test benchmark response validation
+        run: bundle exec ruby bench/test_http_client.rb
+
  test_system:
    runs-on: ubuntu-latest
    permissions:
@@ -109,7 +135,7 @@ jobs:
        options: --health-cmd "redis-cli ping" --health-interval 10s --health-timeout 5s --health-retries 5
    steps:
      - name: Install packages
-        run: sudo apt-get update && sudo apt-get install --no-install-recommends -y libsqlite3-0 libvips curl ffmpeg
+        run: sudo apt-get update && sudo apt-get install --no-install-recommends -y libsqlite3-0 sqlite3 libvips curl ffmpeg

      - name: Checkout code
        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
@@ -117,7 +143,7 @@ jobs:
          persist-credentials: false

      - name: Set up Ruby
-        uses: ruby/setup-ruby@a0102e0972be65f351c307e2d64b9314a57c8073 # v1.324.0
+        uses: ruby/setup-ruby@14594264cd68ce8a2345dd349bc3d138a4ef85c8 # v1.327.0
        env:
          REDIS_URL: redis://localhost:6379/0
        with:
@@ -43,7 +43,6 @@ gem "rqrcode"
 gem "rails_autolink"
 gem "geared_pagination"
 gem "jbuilder"
-gem "net-http-persistent"
 gem "surfguard", github: "basecamp/surfguard" # The SSRF address policy behind RestrictedHTTP
 gem "kredis"
 gem "platform_agent"
@@ -23,16 +23,16 @@ GIT

 GIT
  remote: https://github.com/rails/propshaft.git
-  revision: e49a9de659ff27462015e54dd832e86e762a6ddc
+  revision: dc979db89cd07c72ee4d11d415ae1cb4fd072623
  specs:
-    propshaft (1.2.1)
+    propshaft (1.3.2)
      actionpack (>= 7.0.0)
      activesupport (>= 7.0.0)
      rack

 GIT
  remote: https://github.com/rails/rails.git
-  revision: 1a02651ac37fb64b4de2a2b73461d86acf9c98fb
+  revision: e3d5c569d217c56c022b04acb127a8c24dbe9685
  branch: main
  specs:
    actioncable (8.2.0.alpha)
@@ -63,10 +63,10 @@ GIT
      rack-session (>= 1.0.1)
      rack-test (>= 0.6.3)
      rails-dom-testing (~> 2.2)
-      rails-html-sanitizer (~> 1.6)
+      rails-html-sanitizer (~> 1.7)
      useragent (~> 0.16)
    actiontext (8.2.0.alpha)
-      action_text-trix (~> 2.1.15)
+      action_text-trix (~> 2.1.16)
      actionpack (= 8.2.0.alpha)
      activerecord (= 8.2.0.alpha)
      activestorage (= 8.2.0.alpha)
@@ -77,11 +77,12 @@ GIT
      activesupport (= 8.2.0.alpha)
      builder (~> 3.1)
      erubi (~> 1.11)
+      herb (>= 0.10)
      rails-dom-testing (~> 2.2)
-      rails-html-sanitizer (~> 1.6)
+      rails-html-sanitizer (~> 1.7)
    activejob (8.2.0.alpha)
      activesupport (= 8.2.0.alpha)
-      globalid (>= 0.3.6)
+      globalid (>= 1.4.0)
    activemodel (8.2.0.alpha)
      activesupport (= 8.2.0.alpha)
    activerecord (8.2.0.alpha)
@@ -93,7 +94,7 @@ GIT
      activejob (= 8.2.0.alpha)
      activerecord (= 8.2.0.alpha)
      activesupport (= 8.2.0.alpha)
-      marcel (~> 1.0)
+      marcel (~> 2.0)
    activesupport (8.2.0.alpha)
      base64
      bigdecimal
@@ -104,6 +105,8 @@ GIT
      json
      logger (>= 1.4.2)
      minitest (>= 5.1)
+      psych (>= 4)
+      ractor-dispatch (>= 0.3.0)
      securerandom (>= 0.3)
      tzinfo (~> 2.0, >= 2.0.5)
      uri (>= 0.13.1)
@@ -142,7 +145,7 @@ GEM
    base64 (0.3.0)
    bcrypt (3.1.22)
    benchmark (0.5.0)
-    bigdecimal (3.3.1)
+    bigdecimal (4.1.3)
    brakeman (8.0.6)
      racc
    builder (3.3.0)
@@ -159,7 +162,7 @@ GEM
      regexp_parser (>= 1.5, < 3.0)
      xpath (~> 3.2)
    chunky_png (1.4.0)
-    concurrent-ruby (1.3.7)
+    concurrent-ruby (1.3.8)
    connection_pool (2.5.5)
    crack (1.0.0)
      bigdecimal
@@ -170,9 +173,9 @@ GEM
      irb (~> 1.10)
      reline (>= 0.3.8)
    drb (2.2.3)
-    erb (6.0.4)
+    erb (6.0.7)
    erubi (1.13.1)
-    faker (3.5.2)
+    faker (3.8.0)
      i18n (>= 1.8.11, < 2)
    ffi (1.17.2)
    ffi (1.17.2-aarch64-linux-gnu)
@@ -182,17 +185,22 @@ GEM
    geared_pagination (1.2.0)
      activesupport (>= 5.0)
      addressable (>= 2.5.0)
-    globalid (1.3.0)
+    globalid (1.4.0)
      activesupport (>= 6.1)
    hashdiff (1.2.0)
-    i18n (1.14.7)
+    herb (0.11.0-aarch64-linux-gnu)
+    herb (0.11.0-arm64-darwin)
+    herb (0.11.0-x86_64-darwin)
+    herb (0.11.0-x86_64-linux-gnu)
+    i18n (1.15.2)
      concurrent-ruby (~> 1.0)
    image_processing (1.14.0)
      mini_magick (>= 4.9.5, < 6)
      ruby-vips (>= 2.0.17, < 3)
-    io-console (0.8.1)
-    irb (1.15.3)
+    io-console (0.9.4)
+    irb (1.18.0)
      pp (>= 0.6.0)
+      prism (>= 1.3.0)
      rdoc (>= 4.0.0)
      reline (>= 0.4.2)
    jbuilder (2.14.1)
@@ -219,28 +227,28 @@ GEM
      net-imap
      net-pop
      net-smtp
-    marcel (1.1.0)
+    marcel (2.1.0)
    matrix (0.4.3)
    mini_magick (5.3.1)
      logger
    mini_mime (1.1.5)
-    minitest (5.26.2)
+    minitest (6.0.6)
+      drb (~> 2.0)
+      prism (~> 1.5)
    mocha (3.1.0)
      ruby2_keywords (>= 0.0.5)
    mono_logger (1.1.2)
    multi_json (1.17.0)
    mustermann (3.0.4)
      ruby2_keywords (~> 0.0.1)
-    net-http-persistent (4.0.6)
-      connection_pool (~> 2.2, >= 2.2.4)
-    net-imap (0.6.4.1)
+    net-imap (0.6.7)
      date
      net-protocol
    net-pop (0.1.2)
      net-protocol
-    net-protocol (0.2.2)
+    net-protocol (0.4.0)
      timeout
-    net-smtp (0.5.1)
+    net-smtp (0.5.2)
      net-protocol
    nio4r (2.7.5)
    nokogiri (1.19.4-aarch64-linux-gnu)
@@ -260,18 +268,18 @@ GEM
    platform_agent (1.0.1)
      activesupport (>= 5.2.0)
      useragent (~> 0.16.3)
-    pp (0.6.3)
+    pp (0.6.4)
      prettyprint
    prettyprint (0.2.0)
-    prism (1.4.0)
-    psych (5.2.6)
+    prism (1.9.0)
+    psych (5.5.0)
      date
      stringio
    public_suffix (6.0.2)
    puma (7.2.1)
      nio4r (~> 2.0)
    racc (1.8.1)
-    rack (3.2.6)
+    rack (3.2.7)
    rack-protection (4.2.1)
      base64 (>= 0.1.0)
      logger (>= 1.6.0)
@@ -281,8 +289,9 @@ GEM
      rack (>= 3.0.0)
    rack-test (2.2.0)
      rack (>= 1.3)
-    rackup (2.2.1)
+    rackup (2.3.1)
      rack (>= 3)
+    ractor-dispatch (0.3.0)
    rails-dom-testing (2.3.0)
      activesupport (>= 5.0.0)
      minitest
@@ -295,19 +304,24 @@ GEM
      activesupport (> 3.1)
      railties (> 3.1)
    rainbow (3.1.1)
-    rake (13.3.1)
-    rdoc (6.16.1)
+    rake (13.4.2)
+    rbs (4.2.0)
+      logger
+      prism (>= 1.6.0)
+      tsort
+    rdoc (8.1.0)
      erb
-      psych (>= 4.0.0)
+      prism (>= 1.6.0)
+      rbs (>= 4.0.0)
      tsort
    redis (5.4.1)
      redis-client (>= 0.22.0)
-    redis-client (0.25.2)
+    redis-client (0.30.1)
      connection_pool
    redis-namespace (1.11.0)
      redis (>= 4)
    regexp_parser (2.11.2)
-    reline (0.6.3)
+    reline (0.7.0)
      io-console (~> 0.5)
    resque (2.7.0)
      mono_logger (~> 1)
@@ -355,9 +369,9 @@ GEM
      ffi (~> 1.12)
      logger
    ruby2_keywords (0.0.5)
-    rubyzip (3.4.0)
+    rubyzip (3.7.0)
    securerandom (0.4.1)
-    selenium-webdriver (4.35.0)
+    selenium-webdriver (4.49.0)
      base64 (~> 0.2)
      logger (~> 1.4)
      rexml (~> 3.2, >= 3.2.5)
@@ -382,8 +396,8 @@ GEM
    sqlite3 (2.9.6-x86_64-linux-gnu)
    stimulus-rails (1.3.4)
      railties (>= 6.0.0)
-    stringio (3.1.8)
-    thor (1.4.0)
+    stringio (3.2.0)
+    thor (1.5.0)
    thruster (0.1.23-aarch64-linux)
    thruster (0.1.23-arm64-darwin)
    thruster (0.1.23-x86_64-darwin)
@@ -412,7 +426,7 @@ GEM
    websocket-extensions (0.1.5)
    xpath (3.2.0)
      nokogiri (~> 1.8)
-    zeitwerk (2.7.3)
+    zeitwerk (2.8.3)

 PLATFORMS
  aarch64-linux
@@ -435,7 +449,6 @@ DEPENDENCIES
  kredis
  lexxy (~> 0.9.24)
  mocha
-  net-http-persistent
  ostruct
  platform_agent
  propshaft!
@@ -51,18 +51,20 @@ If you'd rather run the Docker image yourself, you can read more about that in t

 ## Other implementations

-Campfire also has implementations in Django, Laravel, Express, Elixir, Go and Rust:
+Campfire also has implementations in Django, Laravel, Express, Elixir, Go, Rust and C:

-| HTTP workload (requests/sec) | Rails | [Django](https://github.com/basecamp/once-campfire-django) | [Laravel](https://github.com/basecamp/once-campfire-laravel) | [Express](https://github.com/basecamp/once-campfire-express) | [Elixir](https://github.com/basecamp/once-campfire-elixir) | [Go](https://github.com/basecamp/once-campfire-go) | [Rust](https://github.com/basecamp/once-campfire-rust) |
-|---|---:|---:|---:|---:|---:|---:|---:|
-| Room page | 241 | 170 | 164 | 559 | 722 | 3,860 | 36,260 |
-| Messages page | 413 | 196 | 175 | 777 | 1,053 | 5,573 | 40,872 |
-| Sidebar | 552 | 615 | 715 | 4,125 | 1,275 | 19,753 | 34,672 |
-| Search | 435 | 315 | 305 | 1,294 | 1,156 | 7,053 | 33,299 |
-| Post a message | 273 | 154 | 137 | 256 | 801 | 4,767 | 6,896 |
+| HTTP workload (requests/sec) | Rails | [Django](https://github.com/basecamp/once-campfire-django) | [Laravel](https://github.com/basecamp/once-campfire-laravel) | [Express](https://github.com/basecamp/once-campfire-express) | [Elixir](https://github.com/basecamp/once-campfire-elixir) | [Go](https://github.com/basecamp/once-campfire-go) | [Rust](https://github.com/basecamp/once-campfire-rust) | [C](https://github.com/basecamp/once-campfire-c) |
+|---|---:|---:|---:|---:|---:|---:|---:|---:|
+| Room page | 230 | 62 | 760 | 2,622 | 942 | 31,673 | 35,484 | 141,834 |
+| Messages page | 402 | 70 | 924 | 3,245 | 1,267 | 30,746 | 40,674 | 151,564 |
+| Sidebar | 468 | 229 | 1,383 | 34,938 | 2,515 | 18,586 | 34,479 | 159,850 |
+| Search | 399 | 118 | 1,135 | 6,613 | 1,814 | 29,765 | 34,432 | 155,456 |
+| Post a message | 248 | 112 | 498 | 2,088 | 1,400 | 9,073 | 8,998 | 7,460 |

-Measured with 16 concurrent clients on an AMD Ryzen AI MAX+ 395,
-with four hardware threads allocated to each app.
+[Shared verification](https://github.com/basecamp/once-campfire-verification) · [Detailed results](https://github.com/basecamp/once-campfire-verification/blob/main/docs/performance-review.md).
+
+Measured with 16 concurrent clients on an AMD Ryzen AI MAX+ 395 with 32 GB RAM,
+with four hardware cores allocated to each app.

 ## Development

@@ -22,6 +22,6 @@ class PresenceChannel < RoomChannel
    end

    def broadcast_read_room
-      ActionCable.server.broadcast "user_#{current_user.id}_reads", { room_id: membership.room_id }
+      ActionCable.server.broadcast "user_#{current_user.id}_reads", { room_id: membership.room_id, at: Time.current.to_f }
    end
 end
@@ -2,7 +2,7 @@ class Accounts::UsersController < ApplicationController
  before_action :ensure_can_administer, :set_user, only: %i[ update destroy ]

  def index
-    set_page_and_extract_portion_from User.active.ordered.without_bots, per_page: 500
+    set_page_and_extract_portion_from User.visible_to(Current.user).ordered.member, per_page: 500
  end

  def update
@@ -3,9 +3,9 @@ class AccountsController < ApplicationController
  before_action :set_account

  def edit
-    users = account_users.ordered.without_bots
-    @administrators, @members = users.partition(&:administrator?)
-    set_page_and_extract_portion_from users, per_page: 500
+    users = User.visible_to(Current.user).ordered
+    @administrators = users.administrator
+    set_page_and_extract_portion_from users.member, per_page: 500
  end

  def update
@@ -21,12 +21,4 @@ class AccountsController < ApplicationController
    def account_params
      params.require(:account).permit(:name, :logo, settings: {})
    end
-
-    def account_users
-      if Current.user.can_administer?
-        User.where(status: [ :active, :banned ])
-      else
-        User.active
-      end
-    end
 end
@@ -8,6 +8,6 @@ module SetCurrentRequest
  end

  def default_url_options
-    { host: Current.request_host, protocol: Current.request_protocol }.compact_blank
+    { host: Current.request_host, port: Current.request&.optional_port, protocol: Current.request_protocol }.compact_blank
  end
 end
@@ -36,7 +36,7 @@ class Messages::BoostsController < ApplicationController

    def broadcast_create
      @boost.broadcast_append_to @boost.message.room, :messages,
-        target: "boosts_message_#{@boost.message.client_message_id}", partial: "messages/boosts/boost", attributes: { maintain_scroll: true }
+        target: helpers.dom_id(@boost.message, :boosts), partial: "messages/boosts/boost", attributes: { maintain_scroll: true }
    end

    def broadcast_remove
@@ -12,7 +12,7 @@ class RoomsController < ApplicationController
  end

  def destroy
-    @room.destroy
+    @room.destroy_later

    broadcast_remove_room
    redirect_to root_url
@@ -20,7 +20,7 @@ class SearchesController < ApplicationController
  private
    def set_messages
      if query.present?
-        @messages = Current.user.reachable_messages.search(query).with_presentation.last_page_of(100)
+        @messages = Message.search_reachable(Current.user, query)
      else
        @messages = Message.none
      end
@@ -1,8 +1,9 @@
 class UnfurlLinksController < ApplicationController
-  def create
-    opengraph = Opengraph::Metadata.from_url(url_param)
+  # The pasted link decides which hosts an unfurl looks up and fetches from, and how quickly they answer.
+  DEADLINE = 10.seconds

-    if opengraph.valid?
+  def create
+    if opengraph = unfurl(url_param)
      render json: opengraph
    else
      head :no_content
@@ -13,4 +14,14 @@ class UnfurlLinksController < ApplicationController
    def url_param
      params.require(:url)
    end
+
+    # One deadline for everything the link leads to: DNS lookups, redirects, the page, the image check.
+    def unfurl(url)
+      Timeout.timeout(DEADLINE) do
+        Opengraph::Metadata.from_url(url).then { |opengraph| opengraph if opengraph.valid? }
+      end
+    rescue Timeout::Error
+      Rails.logger.warn "Gave up unfurling #{url} after #{DEADLINE.inspect}"
+      nil
+    end
 end
@@ -2,8 +2,9 @@ class Users::SidebarsController < ApplicationController
  DIRECT_PLACEHOLDERS = 20

  def show
-    @direct_memberships, @other_memberships = Current.user.memberships.visible.with_ordered_room.partition { |membership| membership.room.direct? }
-    @direct_memberships = @direct_memberships.sort_by { |membership| membership.room.updated_at }.reverse
+    visible_memberships = Current.user.memberships.visible
+    @direct_memberships = visible_memberships.with_direct_rooms
+    @other_memberships  = visible_memberships.with_ordered_room.without_direct_rooms

    @direct_placeholder_users = find_direct_placeholder_users
  end
@@ -17,11 +17,16 @@ class Messages::AttachmentPresentation
    attr_reader :message, :context
    delegate :tag, :link_to, :broadcast_image_tag, :rails_blob_path, :url_for, to: :context

+    # Previews are made when the message is posted, within limits. One that wasn't, because the file was too large or
+    # the preview failed, isn't attempted again from here: its URL would make it on every view.
    def render_preview
-      if message.attachment.video?
+      case
+      when message.attachment.video?
        video_preview_tag
-      else
+      when message.attachment.representation(:thumb).processed?
        lightboxed_image_preview_tag
+      else
+        render_link
      end
    end

@@ -30,11 +35,17 @@ class Messages::AttachmentPresentation

      inline_media_dimension_constraints(width, height) do
        tag.video \
-          src: rails_blob_path(message.attachment), poster: url_for(message.attachment.preview(format: :webp, resize_to_limit: [ Message::THUMBNAIL_MAX_WIDTH, Message::THUMBNAIL_MAX_HEIGHT ])),
+          src: rails_blob_path(message.attachment), poster: video_poster_url,
          controls: true, preload: :none, width: "100%", height: "100%", class: "message__attachment"
      end
    end

+    # A preview is processed once the frame is drawn, but the poster is a variant of that frame, which can fail on its own.
+    def video_poster_url
+      poster = message.attachment.preview(:poster)
+      url_for(poster) if poster.processed? && poster.image.variant(poster.variation).processed?
+    end
+
    def lightboxed_image_preview_tag
      width, height = preview_dimensions

@@ -36,6 +36,7 @@ module MessagesHelper
        controller: "reply",
        user_id: message.creator_id,
        message_id: message.id,
+        client_message_id: message.client_message_id,
        message_timestamp: message_timestamp_milliseconds,
        message_updated_at: message.updated_at.to_fs(:epoch),
        sort_value: message_timestamp_milliseconds,
@@ -46,6 +46,7 @@ export default class extends Controller {
  }

  messageTargetConnected(target) {
+    this.#reconcilePendingMessage(target)
    this.#formatter.format(target, ThreadStyle.thread)
  }

@@ -125,6 +126,25 @@ export default class extends Controller {

  // Internal

+  // The composer inserts a pending message client-side using the client_message_id
+  // as its DOM id. The server broadcast now renders the real message with a DOM id
+  // derived from the record's primary key, so it no longer replaces the pending
+  // element by id. Reconcile by matching the server message's client_message_id and
+  // dropping our own pending placeholder. Only elements flagged data-pending-message
+  // (client-side placeholders, never broadcast) are removed, so a message another
+  // member posted reusing this client_message_id can never displace a real one.
+  #reconcilePendingMessage(target) {
+    const clientMessageId = target.dataset.clientMessageId
+
+    if (target.dataset.messageId && clientMessageId) {
+      const selector = `[data-pending-message][data-client-message-id="${CSS.escape(clientMessageId)}"]`
+
+      this.messagesTarget.querySelectorAll(selector).forEach((pending) => {
+        if (pending !== target) pending.remove()
+      })
+    }
+  }
+
  async #ensureUpToDate() {
    if (!this.#paginator.upToDate) {
      await this.#paginator.resetToLastPage()
@@ -16,7 +16,7 @@ export default class extends Controller {
    })
  }

-  #read = ({ room_id }) => {
-    this.dispatch("read", { detail: { roomId: room_id } })
+  #read = ({ room_id, at }) => {
+    this.dispatch("read", { detail: { roomId: room_id, at } })
  }
 }
@@ -7,8 +7,13 @@ export default class extends Controller {
  static classes = [ "unread" ]

  #disconnected = true
+  #connection = 0
+  #discardDirectEditor = false
+  #readAt = new Map()

  async connect() {
+    this.element.addEventListener("turbo:click", this.#directNavigation)
+    this.element.addEventListener("turbo:before-frame-render", this.#preserveDirectEditor)
    this.channel ??= await cable.subscribeTo({ channel: "UnreadRoomsChannel" }, {
      connected: this.#channelConnected.bind(this),
      disconnected: this.#channelDisconnected.bind(this),
@@ -17,7 +22,11 @@ export default class extends Controller {
  }

  disconnect() {
+    this.element.removeEventListener("turbo:click", this.#directNavigation)
+    this.element.removeEventListener("turbo:before-frame-render", this.#preserveDirectEditor)
+    this.#discardDirectEditor = false
    ignoringBriefDisconnects(this.element, () => {
+      this.#channelDisconnected()
      this.channel?.unsubscribe()
      this.channel = null
    })
@@ -27,19 +36,58 @@ export default class extends Controller {
    this.read({ detail: { roomId: Current.room.id } })
  }

-  read({ detail: { roomId } }) {
+  read({ detail: { roomId, at } }) {
    const room = this.#findRoomTarget(roomId)

+    if (at) {
+      this.#readAt.set(Number(roomId), Math.max(Number(at), this.#readAt.get(Number(roomId)) ?? 0))
+    }
+
    if (room) {
      room.classList.remove(this.unreadClass)
      this.dispatch("read", { detail: { targetId: roomId } })
    }
  }

-  #channelConnected() {
+  // Preserve the live picker through background refreshes, but allow deliberate navigation.
+  #directNavigation = event => {
+    if (event.target.closest("a")?.dataset.turboFrame === this.element.id) {
+      this.#discardDirectEditor = true
+    }
+  }
+
+  #preserveDirectEditor = event => {
+    if (event.target !== this.element) return
+    const discard = this.#discardDirectEditor
+    this.#discardDirectEditor = false
+    if (discard) return
+
+    const current = this.element.querySelector("#direct_rooms_control")
+    const incoming = event.detail.newFrame.querySelector("#direct_rooms_control")
+    if (!current?.querySelector('[data-autocomplete-target="input"]') || !incoming) return
+
+    // Turbo's Bardo moves the actual element, keeping input, selections and controllers.
+    const controls = [ current, incoming ].map(element => [ element, element.hasAttribute("data-turbo-permanent") ])
+    for (const [ element ] of controls) element.setAttribute("data-turbo-permanent", "")
+    const render = event.detail.render
+    event.detail.render = (...args) => {
+      try {
+        return render(...args)
+      } finally {
+        for (const [ element, permanent ] of controls) element.toggleAttribute("data-turbo-permanent", permanent)
+      }
+    }
+  }
+
+  async #channelConnected() {
    if (this.#disconnected) {
      this.#disconnected = false
-      this.element.reload()
+      const connection = ++this.#connection
+      // Reloading an unfinished frame aborts its response body reader.
+      await this.element.loaded
+      if (this.element.isConnected && !this.#disconnected && connection === this.#connection) {
+        this.element.reload()
+      }
    }
  }

@@ -47,11 +95,11 @@ export default class extends Controller {
    this.#disconnected = true
  }

-  #unread({ roomId }) {
+  #unread({ roomId, at }) {
    const unreadRoom = this.#findRoomTarget(roomId)

    if (unreadRoom) {
-      if (Current.room.id != roomId) {
+      if (Current.room.id != roomId && !this.#readSince(roomId, at)) {
        unreadRoom.classList.add(this.unreadClass)
      }

@@ -59,6 +107,12 @@ export default class extends Controller {
    }
  }

+  // Notices fan out one member at a time, so one can arrive after the member has already
+  // read the room in another tab. It still reorders the room, but doesn't mark it unread.
+  #readSince(roomId, at) {
+    return Number(at) <= this.#readAt.get(Number(roomId))
+  }
+
  #findRoomTarget(roomId) {
    return this.roomTargets.find(roomTarget => roomTarget.dataset.roomId == roomId)
  }
@@ -0,0 +1,5 @@
+class Message::BroadcastUnreadRoomJob < ApplicationJob
+  def perform(message)
+    message.broadcast_unread_room
+  end
+end
@@ -0,0 +1,8 @@
+class Room::DestroyJob < ApplicationJob
+  discard_on ActiveJob::DeserializationError
+
+  def perform(room, former_member_ids)
+    room.reset_remote_connections_of(former_member_ids)
+    room.destroy_one_message_at_a_time
+  end
+end
@@ -9,7 +9,8 @@ class Membership < ApplicationRecord
  enum :involvement, %w[ invisible nothing mentions everything ].index_by(&:itself), prefix: :involved_in

  scope :with_ordered_room, -> { includes(:room).joins(:room).order("LOWER(rooms.name)") }
-  scope :without_direct_rooms, -> { joins(:room).where.not(room: { type: "Rooms::Direct" }) }
+  scope :with_direct_rooms, -> { includes(:room).joins(:room).merge(Room.directs).order(rooms: { updated_at: :desc }) }
+  scope :without_direct_rooms, -> { joins(:room).merge(Room.without_directs) }

  scope :visible, -> { where.not(involvement: :invisible) }
  scope :unread,  -> { where.not(unread_at: nil) }
@@ -24,10 +24,6 @@ class Message < ApplicationRecord
    body.to_plain_text.presence || attachment&.filename&.to_s || ""
  end

-  def to_key
-    [ client_message_id ]
-  end
-
  def content_type
    case
    when attachment?    then "attachment"
@@ -4,9 +4,14 @@ module Message::Attachment
  THUMBNAIL_MAX_WIDTH = 1200
  THUMBNAIL_MAX_HEIGHT = 800

+  # Decoding a picture, or a video's frame, costs in proportion to its pixels, however small the file. The largest
+  # phone photos have 200 million.
+  THUMBNAIL_MAX_PIXELS = 250_000_000
+
  included do
    has_one_attached :attachment do |attachable|
      attachable.variant :thumb, resize_to_limit: [ THUMBNAIL_MAX_WIDTH, THUMBNAIL_MAX_HEIGHT ]
+      attachable.variant :poster, format: :webp, resize_to_limit: [ THUMBNAIL_MAX_WIDTH, THUMBNAIL_MAX_HEIGHT ]
    end
  end

@@ -30,12 +35,25 @@ module Message::Attachment
      attachment&.analyze
    end

+    # A file that ffmpeg or libvips can't decode is still the message: post it without a preview.
    def process_attachment_thumbnail
+      return if too_many_pixels_to_preview?
+
      case
      when attachment.video?
-        attachment.preview(format: :webp).processed
+        attachment.preview(:poster).processed
      when attachment.representable?
        attachment.representation(:thumb).processed
      end
+    rescue ActiveStorage::PreviewError, Vips::Error => error
+      Rails.logger.warn "Posted #{attachment.filename} without a preview: #{error.class}: #{error.message.lines.first&.chomp}"
+    end
+
+    # Without a size, the analyzer couldn't read the file's header, and the previewer wouldn't either.
+    def too_many_pixels_to_preview?
+      if attachment.image? || attachment.video?
+        width, height = attachment.metadata.values_at(:width, :height)
+        width.nil? || height.nil? || width * height > THUMBNAIL_MAX_PIXELS
+      end
    end
 end
@@ -1,21 +1,29 @@
 module Message::Broadcasts
  def broadcast_create
    broadcast_append_to room, :messages, target: [ room, :messages ]
-    broadcast_unread_room
+    broadcast_unread_room_later
  end

  def broadcast_remove
    broadcast_remove_to room, :messages
  end

-  private
-    # Fanned out to the room's members rather than published on one global stream, so
-    # that the timing of activity in a room only reaches people who are in it.
-    def broadcast_unread_room
-      payload = ActiveSupport::JSON.encode(roomId: room.id)
+  # Fanned out to the room's members rather than published on one global stream, so
+  # that the timing of activity in a room only reaches people who are in it. Of those,
+  # only the ones who still have it unread or are in it now: by the time this runs,
+  # someone may have opened the room and moved on, and would see it marked unread again.
+  def broadcast_unread_room
+    payload = ActiveSupport::JSON.encode(roomId: room.id, at: created_at.to_f)

-      room.memberships.pluck(:user_id).each do |user_id|
-        ActionCable.server.broadcast UnreadRoomsChannel.stream_name_for(user_id), payload, coder: nil
-      end
+    room.memberships.unread.or(room.memberships.connected).pluck(:user_id).each do |user_id|
+      ActionCable.server.broadcast UnreadRoomsChannel.stream_name_for(user_id), payload, coder: nil
+    end
+  end
+
+  private
+    # The fanout is one publish per member, which in a big room takes far longer than the
+    # rest of posting a message, so it runs in a job rather than while the poster waits.
+    def broadcast_unread_room_later
+      Message::BroadcastUnreadRoomJob.perform_later(self)
    end
 end
@@ -36,7 +36,10 @@ module Message::Pagination
    scope :page_after, ->(message) { after(message).first_page }

    scope :page_created_since, ->(time) { where("created_at > ?", time).first_page }
-    scope :page_updated_since, ->(time) { where("updated_at > ?", time).last_page }
+    # Sorting on +created_at, which no index covers, has SQLite find the few messages updated since
+    # through (room_id, updated_at) and sort just those, instead of walking the whole room by
+    # (room_id, created_at) looking for them.
+    scope :page_updated_since, ->(time) { Page.load(where("updated_at > ?", time).reorder(Arel.sql("+messages.created_at")), :last, PAGE_SIZE) }
  end

  class_methods do
@@ -49,7 +52,7 @@ module Message::Pagination
    end

    def paged?
-      count > PAGE_SIZE
+      offset(PAGE_SIZE).exists?
    end
  end
 end
@@ -3,7 +3,8 @@ module Message::Searchable

  included do
    after_create_commit  :create_in_index
-    after_update_commit  :update_in_index
+    before_update        -> { @attachment_replaced = attachment_changes.key?("attachment") }
+    after_update_commit  :update_in_index, if: :indexed_text_changed?
    after_destroy_commit :remove_from_index

    scope :search, ->(query) { joins("join message_search_index idx on messages.id = idx.rowid").where("idx.body match ?", match_terms(query)).ordered }
@@ -17,11 +18,43 @@ module Message::Searchable
    end
  end

+  class_methods do
+    def search_reachable(user, query, size: 100)
+      relation = user.reachable_messages.search(query).with_presentation
+      probe = connection.select_rows(sanitize_sql([ <<~SQL, user.id, match_terms(query) ]))
+        SELECT m.id, mm.user_id IS NOT NULL FROM message_search_index idx
+        JOIN messages m ON m.id = idx.rowid
+        LEFT JOIN memberships mm ON mm.room_id = m.room_id AND mm.user_id = ?
+        WHERE idx.body MATCH ? ORDER BY idx.rowid DESC LIMIT 1000
+      SQL
+      ids = probe.filter_map { |id, reachable| id if reachable == 1 }.first(size)
+
+      # A scoped fallback preserves older reachable hits behind a large private history.
+      if ids.size < size && probe.size == 1000
+        relation.last_page_of_matches(size)
+      else
+        Message::Pagination::Page.load(relation.where(id: ids).reorder(:id), :last, size)
+      end
+    end
+
+    # Orders by the index's rowid, which is the message id, so SQLite walks the full-text index
+    # newest first and stops at the page. Ordering by created_at sorted every match before paging.
+    def last_page_of_matches(size)
+      Message::Pagination::Page.load(reorder("idx.rowid"), :last, size)
+    end
+  end
+
  private
    def create_in_index
      execute_sql_with_binds "insert into message_search_index(rowid, body) values (?, ?)", id, plain_text_body
    end

+    # Boosts touch their message too: rewrite the index only when the body was saved or the attachment replaced.
+    # Active Storage clears attachment_changes in its own after_commit, before this one runs.
+    def indexed_text_changed?
+      association(:rich_text_body).target&.saved_change_to_body? || @attachment_replaced
+    end
+
    def update_in_index
      execute_sql_with_binds "update message_search_index set body = ? where rowid = ?", plain_text_body, id
    end
@@ -5,6 +5,7 @@ class Opengraph::Fetch
  ALLOWED_DOCUMENT_CONTENT_TYPE = "text/html"
  MAX_BODY_SIZE = 5.megabytes
  MAX_REDIRECTS = 10
+  TIMEOUT = 7.seconds # to connect (TLS included), and for each read and write, as Webhook does

  class TooManyRedirectsError < StandardError; end
  class RedirectDeniedError < StandardError; end
@@ -22,9 +23,11 @@ class Opengraph::Fetch
  end

  private
+    # The timeouts bound each operation. A host can still send its headers or body a byte at a time,
+    # in as many reads as it likes: UnfurlLinksController puts the whole unfurl under one deadline.
    def request(url, request_class, ip:)
      MAX_REDIRECTS.times do
-        Net::HTTP.start(url.host, url.port, ipaddr: ip, use_ssl: url.scheme == "https") do |http|
+        Net::HTTP.start(url.host, url.port, ipaddr: ip, use_ssl: url.scheme == "https", **timeouts) do |http|
          http.request request_class.new(url) do |response|
            if response.is_a?(Net::HTTPRedirection)
              url, ip = resolve_redirect(response["location"])
@@ -38,6 +41,11 @@ class Opengraph::Fetch
      raise TooManyRedirectsError
    end

+    # Without max_retries: 0, Net::HTTP sends a GET or HEAD that timed out a second time.
+    def timeouts
+      { open_timeout: TIMEOUT, read_timeout: TIMEOUT, write_timeout: TIMEOUT, max_retries: 0 }
+    end
+
    def resolve_redirect(location)
      url = URI.parse(location)
      raise RedirectDeniedError unless url.is_a?(URI::HTTP)
@@ -14,9 +14,9 @@ class Push::Subscription < ApplicationRecord
  validates :endpoint, presence: true
  validate :validate_endpoint_url

-  def notification(**params)
+  def notification(badge: user.memberships.unread.count, **params)
    # Defer DNS lookup to the delivery worker to prevent rebinding
-    WebPush::Notification.new(**params, badge: user.memberships.unread.count, endpoint: endpoint, endpoint_ip_resolver: method(:resolved_endpoint_ip), p256dh_key: p256dh_key, auth_key: auth_key)
+    WebPush::Notification.new(**params, badge: badge, endpoint: endpoint, endpoint_ip_resolver: method(:resolved_endpoint_ip), p256dh_key: p256dh_key, auth_key: auth_key)
  end

  # Validate at point of use, not just when saved.
@@ -45,6 +45,34 @@ class Room < ApplicationRecord
    end
  end

+  # Takes the room away from its members at once, and leaves its messages to Room::DestroyJob. Destroying them in
+  # the room's own transaction held the database's write lock, and stopped every other write, until the last one.
+  # An open room is closed first, so that no one who joins the account before the job runs is let in.
+  def destroy_later
+    room = open? ? becomes!(Rooms::Closed) : self
+
+    former_member_ids = transaction do
+      room.save!
+      room.memberships.pluck(:user_id).tap { room.memberships.delete_all }
+    end
+
+    Room::DestroyJob.perform_later(room, former_member_ids)
+  end
+
+  # Deleting the memberships skips the reset that revoking one does, so the former members get it here, all at once,
+  # before the messages go: their connections stop receiving the room's streams. It's done in the job, not the
+  # request, because it costs a Redis round trip per member.
+  def reset_remote_connections_of(user_ids)
+    User.where(id: user_ids).find_each(&:reset_remote_connections)
+  end
+
+  # Each message is destroyed in its own transaction, so other writes get through in between. Any message posted
+  # meanwhile goes with the room.
+  def destroy_one_message_at_a_time
+    messages.find_each(&:destroy)
+    destroy
+  end
+
  def receive(message)
    unread_memberships(message)
    push_later(message)
@@ -53,9 +53,12 @@ class Room::MessagePusher
      relevant_subscriptions.merge(Membership.involved_in_mentions).where(user_id: mentionees.ids)
    end

+    # Banning keeps the user's subscriptions, so that unbanning brings their notifications back,
+    # but nobody who can't sign in should be sent what's said in their rooms meanwhile.
    def relevant_subscriptions
      Push::Subscription
        .joins(user: :memberships)
+        .merge(User.active)
        .merge(Membership.visible.disconnected.where(room: room).where.not(user: message.creator))
    end

@@ -7,12 +7,15 @@ class Rooms::Direct < Room
    end

    private
-      # FIXME: Find a more performant algorithm that won't be a problem on accounts with 10K+ direct rooms,
-      # which could be to store the membership id list as a hash on the room, and use that for lookup.
+      # Among the first user's rooms, the one whose members are exactly these users: as many
+      # memberships as users, and all of them theirs.
      def find_for(users)
-        all.joins(:users).detect do |room|
-          Set.new(room.user_ids) == Set.new(users.pluck(:id))
-        end
+        user_ids = users.pluck(:id).uniq
+
+        where(id: Membership.where(user_id: user_ids.first).select(:room_id))
+          .joins(:memberships).group(:id)
+          .having("COUNT(*) = :size AND COUNT(CASE WHEN memberships.user_id IN (:user_ids) THEN 1 END) = :size", size: user_ids.size, user_ids: user_ids)
+          .first
      end
  end

@@ -50,8 +50,18 @@ class User < ApplicationRecord
  end

  private
+    # One statement, which SQLite runs under the write lock, so no room is closed between reading the open ones and
+    # inserting. A room that Room#destroy_later closes is either closed before, and skipped, or after, and takes this
+    # membership away with the others. A room that granted itself to every user when it was created is skipped too,
+    # as insert_all did.
    def grant_membership_to_open_rooms
-      Membership.insert_all(Rooms::Open.pluck(:id).collect { |room_id| { room_id: room_id, user_id: id } })
+      now = Time.current
+
+      Membership.connection.execute Membership.sanitize_sql([ <<~SQL, id, now, now ])
+        insert into memberships (room_id, user_id, created_at, updated_at)
+        select id, ?, ?, ? from rooms where type = 'Rooms::Open'
+        on conflict do nothing
+      SQL
    end

    def deactived_email_address
@@ -1,6 +1,11 @@
 module User::Bannable
  extend ActiveSupport::Concern

+  included do
+    # Administrators still see banned users, so they can lift the ban.
+    scope :visible_to, ->(user) { user.can_administer? ? where(status: %i[ active banned ]) : active }
+  end
+
  def ban
    transaction do
      create_bans_from_sessions
@@ -11,6 +11,11 @@ module User::Mentionable
    "users/mention"
  end

+  # How a mention appears inside the editor, matching the prompt's editor template.
+  def to_editor_content_attachment_partial_path
+    "users/mention"
+  end
+
  def attachable_plain_text_representation(caption)
    "@#{name}"
  end
@@ -63,7 +63,7 @@ class Webhook < ApplicationRecord
    end

    def extract_attachment_from(response)
-      if response.content_type && mime_type = Mime::Type.lookup(response.content_type)
+      if response.code == "200" && response.content_type && mime_type = Mime::Type.lookup(response.content_type)
        ActiveStorage::Blob.create_and_upload! \
          io: StringIO.new(response.body), filename: "attachment.#{mime_type.symbol}", content_type: mime_type.to_s
      end
@@ -76,10 +76,9 @@
        <% end %>

        <label class="switch">
-          <input type="checkbox"
-                class="switch__input"
-                <%= "checked" if Current.account.settings.restrict_room_creation_to_administrators? %>
-                data-action="change->form#submit">
+          <%= tag.input type: "checkbox", class: "switch__input",
+                checked: Current.account.settings.restrict_room_creation_to_administrators?,
+                data: { action: "change->form#submit" } %>
          <span class="switch__btn round"></span>
          <span class="for-screen-reader">
            Must be admin to create new rooms
@@ -102,11 +101,11 @@
      <turbo-frame id="account_users">
        <%= render partial: "accounts/users/user", collection: @administrators, as: :user %>

-        <% if @administrators.any? && @members.any? %>
+        <% if @administrators.any? && @page.records.any? %>
          <hr class="separator full-width" style="--border-style: solid">
        <% end %>

-        <%= render partial: "accounts/users/user", collection: @members, as: :user %>
+        <%= render partial: "accounts/users/user", collection: @page.records, as: :user %>
        <%= render "accounts/users/next_page_container", page: @page.next_param unless @page.last? %>
      </turbo-frame>
    </menu>
@@ -0,0 +1,13 @@
+<%# Files are posted as a message's own attachment, which gets its preview when it's posted; the composer adds none to
+    the rich text. One that's there anyway is shown by its name, with no preview: Action Text's partial links to a URL
+    that makes one on view, and nothing else in the app would make it. %>
+<figure class="attachment attachment--file attachment--<%= blob.filename.extension %>">
+  <figcaption class="attachment__caption">
+    <% if caption = blob.try(:caption) %>
+      <%= caption %>
+    <% else %>
+      <span class="attachment__name"><%= blob.filename %></span>
+      <span class="attachment__size"><%= number_to_human_size blob.byte_size %></span>
+    <% end %>
+  </figcaption>
+</figure>
@@ -40,9 +40,9 @@
      <div class="flash" data-controller="element-removal" data-action="animationend->element-removal#remove">
        <div class="flash__inner shadow" style="<%= "--flash-background: var(--color-negative)" if flash[:alert] %>">
          <% if flash[:alert] %>
-            <%= image_tag "alert.svg", aria: { hidden: true }, size: 24, class: "colorize--white" %></span>
+            <%= image_tag "alert.svg", aria: { hidden: true }, size: 24, class: "colorize--white" %>
          <% else %>
-            <%= image_tag "check.svg", aria: { hidden: true }, size: 24, class: "colorize--white" %></span>
+            <%= image_tag "check.svg", aria: { hidden: true }, size: 24, class: "colorize--white" %>
          <% end %>
        </div>
        <span class="for-screen-reader" role="alert" aria-atomic="true"><%= notice %></span>
@@ -1,7 +1,7 @@
 <%# Be sure to check/update messages/_template.html.erb when changing this file %>

 <%# Bump this version when the message presentation filters change what they emit. Editing this line changes the template digest, which busts BOTH this fragment cache and the collection cache that keys on this partial's digest (helper Ruby changes alone don't). %>
-<% cache [ message, "presentation-v3" ] do %>
+<% cache [ message, "presentation-v6" ] do %>
  <%= message_tag message do %>
    <h2 class="message__day-separator"><%= local_datetime_tag message.created_at, style: :date %></h2>

@@ -2,6 +2,8 @@
  <div class="message message--me $messageClasses$"
      id="message_$clientMessageId$"
      data-format-message-target="message"
+      data-pending-message="true"
+      data-client-message-id="$clientMessageId$"
      data-user-id="<%= Current.user.id %>"
      data-message-timestamp="$messageTimestamp$"
      data-messages-target="message">
@@ -1,20 +1,18 @@
-<% cache boost do %>
-  <div id="<%= dom_id(boost) %>"
-      class="boost boost-item flex-inline postion--relative max-width align-center fill-white gap"
-      data-controller="boost-delete" data-boost-delete-perform-class="boost--deleting" data-boost-delete-reveal-class="expanded" data-boost-delete-booster-id-value="<%= boost.booster.id %>">
-    <figure class="avatar boost__avatar flex-item-no-shrink">
-      <%= avatar_tag boost.booster, aria: { label: "#{boost.booster.name} boosted #{boost.content}" } %>
-    </figure>
+<div id="<%= dom_id(boost) %>"
+    class="boost boost-item flex-inline postion--relative max-width align-center fill-white gap"
+    data-controller="boost-delete" data-boost-delete-perform-class="boost--deleting" data-boost-delete-reveal-class="expanded" data-boost-delete-booster-id-value="<%= boost.booster.id %>">
+  <figure class="avatar boost__avatar flex-item-no-shrink">
+    <%= avatar_tag boost.booster, aria: { label: "#{boost.booster.name} boosted #{boost.content}" } %>
+  </figure>

-    <%= tag.span boost.content, role: "button",
-          class: [ "txt-small", { "txt-medium": boost.content.all_emoji? } ],
-          data: { action: "click->boost-delete#reveal keydown.enter->boost-delete#reveal:prevent", boost_delete_target: "content" } %>
+  <%= tag.span boost.content, role: "button",
+        class: [ "txt-small", { "txt-medium": boost.content.all_emoji? } ],
+        data: { action: "click->boost-delete#reveal keydown.enter->boost-delete#reveal:prevent", boost_delete_target: "content" } %>

-    <%= button_to message_boost_path(boost.message, boost), method: :delete, data: { action: "boost-delete#perform", boost_delete_target: "button" },
-          class: "btn btn--negative flex-item-justify-end boost__delete" do %>
-      <%= image_tag "minus.svg", size: 20, aria: { hidden: "true" } %>
-      <span class="for-screen-reader">Delete this boost</span>
-    <% end %>
-  </div>
-  <span id="delete_boost_accessible_label" class="for-screen-reader">Press enter to delete this boost</span>
-<% end %>
+  <%= button_to message_boost_path(boost.message, boost), method: :delete, data: { action: "boost-delete#perform", boost_delete_target: "button" },
+        class: "btn btn--negative flex-item-justify-end boost__delete" do %>
+    <%= image_tag "minus.svg", size: 20, aria: { hidden: "true" } %>
+    <span class="for-screen-reader">Delete this boost</span>
+  <% end %>
+</div>
+<span id="delete_boost_accessible_label" class="for-screen-reader">Press enter to delete this boost</span>
@@ -2,7 +2,7 @@
  <div class="boosts flex flex-wrap align-center gap full-width" style="--column-gap: 0.4ch; --row-gap: 0"
      data-controller="turbo-streaming" data-action="turbo:submit-start->turbo-streaming#unsubscribe">
    <div class="flex-inline flex-wrap gap" id="<%= dom_id(message, :boosts) %>" data-turbo-streaming-target="container">
-      <%= render partial: "messages/boosts/boost", collection: message.boosts.ordered, cached: true %>
+      <%= render partial: "messages/boosts/boost", collection: message.boosts.loaded? ? message.boosts.sort_by(&:created_at) : message.boosts.ordered.includes(:booster) %>
    </div>

    <%= turbo_frame_tag message, :new_boost do %>
@@ -6,8 +6,8 @@
      <%= image_tag "disclosure.svg", aria: { hidden: "true" }, size: 10, class: "disclosure" %>
    </summary>

-    <% case
-      when platform.firefox? && platform.android? %>
+    <% case %>
+      <% when platform.firefox? && platform.android? %>
        <ol>
          <li>Tap <em><%= image_tag "lock.svg", alt: "the View site information button", size: 20 %></em> in the address bar.</li>
          <li>Tap <em>Notification</em> to change to <em>Allowed</em>.</li>
@@ -6,10 +6,10 @@
      <%= image_tag "disclosure.svg", aria: { hidden: "true" }, size: 10, class: "disclosure" %>
    </summary>

-    <% case
-      when platform.edge? %>
+    <% case %>
+      <% when platform.edge? %>
        <ol>
-          <li>Click <em><%= image_tag "install-edge.svg", alt: "the app available - install Campfire chat button", size: 16 %></em>in the address bar.</li>
+          <li>Click <em><%= image_tag "external/install-edge.svg", alt: "the app available - install Campfire chat button", size: 16 %></em>in the address bar.</li>
          <li>Click <em>Install</em>.</li>
        </ol>
      <% when platform.chrome? && platform.android? %>
@@ -5,8 +5,8 @@
    <%= image_tag "disclosure.svg", aria: { hidden: "true" }, size: 10, class: "disclosure" %>
  </summary>

-  <% case
-    when platform.firefox? && platform.android? %>
+  <% case %>
+    <% when platform.firefox? && platform.android? %>
      <ol>
        <li>Tap the <em><%= image_tag "menu-dots-vertical.svg", alt: "More options", size: 16 %></em> menu button.</li>
        <li>Tap <em>Settings</em>.</li>
@@ -1 +1,2 @@
-<%= auto_submit_form_with method: :put %>
+<%= auto_submit_form_with method: :put do %>
+<% end %>
@@ -22,5 +22,6 @@ CSRF disabling. Unread fanout excludes adapter I/O.
 The HTTP driver uses production Puma/Redis with one worker and five threads. Ruby threads
 each maintain a keep-alive connection, request uncompressed responses, and consume the
 whole body. Login uses normal CSRF protection; all warmup and measured responses must be
-HTTP 200 without transport errors. Measurements exclude Thruster, TLS and gzip. Client
+HTTP 200 without transport errors, complete HTML, exact independently seeded message windows
+and the visible sidebar rooms; this applies to every warmup and measured response. Measurements exclude Thruster, TLS and gzip. Client
 CPU, JIT warmup and GC can affect throughput; repeat runs and check client saturation.
@@ -3,6 +3,7 @@
 require "socket"
 require_relative "support"
 require_relative "http_client"
+require_relative "response_contract"

 include BenchmarkSupport
 options = parse_options("Compare HTTP throughput with Ruby keep-alive clients; every response must be HTTP 200.",
@@ -58,9 +59,10 @@ begin
      cookie = client.login(labels)
      results = {}
      paths.each do |name, path|
-        client.measure(path, cookie, concurrency: 1, duration: 3)
+        contract = BenchmarkResponseContract.new(name, File.join(data, "storage/db/production.sqlite3"), labels)
+        client.measure(path, cookie, concurrency: 1, duration: 3, contract: contract)
        concurrencies.each do |concurrency|
-          results["#{name}_#{concurrency}"] = client.measure(path, cookie, concurrency: concurrency, duration: options[:duration])
+          results["#{name}_#{concurrency}"] = client.measure(path, cookie, concurrency: concurrency, duration: options[:duration], contract: contract)
        end
      end
      write_json(File.join(options[:output], "#{side}-#{iteration + 1}.json"), results)
@@ -33,12 +33,12 @@ class BenchmarkHTTPClient
    cookie_header(cookies)
  end

-  def measure(path, cookie, concurrency:, duration:)
+  def measure(path, cookie, concurrency:, duration:, contract:)
    start = clock
    deadline = start + duration
    workers = Array.new(concurrency) do
      Thread.new do
-        result = { latencies: [], statuses: Hash.new(0), bytes: 0, errors: 0 }
+        result = { latencies: [], statuses: Hash.new(0), bytes: 0, errors: 0, invalid_responses: 0 }
        while clock < deadline
          begin
            connection.start do |http|
@@ -48,6 +48,7 @@ class BenchmarkHTTPClient
                result[:latencies] << (clock - requested) * 1000
                result[:statuses][response.code] += 1
                result[:bytes] += response.body.bytesize
+                result[:invalid_responses] += 1 unless contract.valid?(response)
              end
            end
          rescue IOError, SystemCallError, Timeout::Error, SocketError, Net::HTTPBadResponse
@@ -64,9 +65,10 @@ class BenchmarkHTTPClient
    statuses = Hash.new(0)
    samples.each { |sample| sample[:statuses].each { |status, count| statuses[status] += count } }
    errors = samples.sum { |sample| sample[:errors] }
-    raise "#{path}: HTTP statuses #{statuses}, #{errors} transport errors" unless errors.zero? && statuses.keys == [ "200" ]
+    invalid = samples.sum { |sample| sample[:invalid_responses] }
+    raise "#{path}: HTTP statuses #{statuses}, #{errors} transport errors, #{invalid} invalid bodies" unless errors.zero? && invalid.zero? && statuses.keys == [ "200" ]
    { path: path, conc: concurrency, gzip: false, secs: elapsed, rps: latencies.size / elapsed,
-      ok: latencies.size, statuses: statuses, errors: errors,
+      ok: latencies.size, statuses: statuses, errors: errors, invalid_responses: invalid, validation: "route-contract-v1",
      avg_bytes: samples.sum { |sample| sample[:bytes] } / latencies.size,
      latency_ms: { p50: percentile(latencies, 0.50), p95: percentile(latencies, 0.95), p99: percentile(latencies, 0.99) } }
  end
@@ -0,0 +1,43 @@
+require "json"
+require "open3"
+require "cgi"
+
+# Independent seed SQL determines the required result window before any response is sampled.
+class BenchmarkResponseContract
+  def initialize(kind, database, labels)
+    @kind = kind
+    room = Integer(labels.fetch("rooms.watercooler"))
+    user = labels.fetch("emails.david").gsub("'", "''")
+    @ids = case kind
+    when "room"
+      query(database, "SELECT id FROM messages WHERE room_id=#{room} ORDER BY created_at DESC LIMIT 40").reverse.map { |row| row.fetch("id") }
+    when "messages"
+      before = Integer(labels.fetch("messages.busy_060"))
+      query(database, "SELECT id FROM messages WHERE room_id=#{room} AND created_at<(SELECT created_at FROM messages WHERE id=#{before}) ORDER BY created_at DESC LIMIT 40").reverse.map { |row| row.fetch("id") }
+    when "search"
+      query(database, "SELECT m.id FROM message_search_index idx JOIN messages m ON m.id=idx.rowid JOIN memberships mm ON mm.room_id=m.room_id JOIN users u ON u.id=mm.user_id WHERE u.email_address='#{user}' AND idx.body MATCH 'coffee' ORDER BY m.id DESC LIMIT 100").reverse.map { |row| row.fetch("id") }
+    when "sidebar"
+      @names = query(database, "SELECT r.name FROM rooms r JOIN memberships mm ON mm.room_id=r.id JOIN users u ON u.id=mm.user_id WHERE u.email_address='#{user}' AND r.type='Rooms::Open' AND mm.involvement<>'invisible'").map { |row| CGI.escapeHTML(row.fetch("name")) }
+      nil
+    else
+      raise "Unknown route contract #{kind}"
+    end
+  end
+
+  def valid?(response)
+    body = response.body.to_s
+    return false unless response.code == "200" && response["content-type"].to_s.split(";").first == "text/html" &&
+      [ nil, "identity" ].include?(response["content-encoding"]) && !body.empty? && body.dup.force_encoding("UTF-8").valid_encoding?
+    return false if @kind != "messages" && !(body.lstrip.start_with?("<!DOCTYPE html>") && body.rstrip.end_with?("</html>"))
+    return false if @ids && body.scan(/data-message-id="(\d+)"/).flatten.map(&:to_i) != @ids
+    return false if @names && (!body.include?("shared_rooms") || @names.any? { |name| !body.include?(name) })
+    true
+  end
+
+  private
+    def query(database, sql)
+      output, errors, status = Open3.capture3("sqlite3", "-readonly", "-json", database, sql)
+      raise "Cannot read benchmark seed: #{errors}" unless status.success?
+      JSON.parse(output)
+    end
+end
@@ -0,0 +1,71 @@
+require "minitest/autorun"
+require "fileutils"
+require "socket"
+require "sqlite3"
+require "tmpdir"
+require_relative "http_client"
+require_relative "response_contract"
+
+class BenchmarkHTTPClientTest < Minitest::Test
+  def setup
+    @directory = Dir.mktmpdir("campfire-benchmark-test")
+    database = File.join(@directory, "seed.sqlite3")
+    SQLite3::Database.new(database).tap do |db|
+      db.execute("CREATE TABLE messages (id INTEGER PRIMARY KEY, room_id INTEGER, created_at TEXT)")
+      db.execute("INSERT INTO messages VALUES (42, 1, '2026-10-01'), (43, 1, '2026-10-02')")
+      db.close
+    end
+    @contract = BenchmarkResponseContract.new("room", database, "rooms.watercooler" => 1, "emails.david" => "david@example.com")
+    @body = '<!DOCTYPE html><html><article data-message-id="42">coffee</article><article data-message-id="43">meeting</article></html>'
+  end
+
+  def teardown
+    FileUtils.remove_entry(@directory)
+  end
+
+  def test_seeded_contract_rejects_error_pages_truncation_wrong_windows_and_encoding
+    assert @contract.valid?(response(@body))
+    refute @contract.valid?(response("Something went wrong"))
+    refute @contract.valid?(response(@body.delete_suffix("</html>")))
+    refute @contract.valid?(response(@body.sub('data-message-id="42"', 'data-message-id="99"')))
+    refute @contract.valid?(response(@body, "text/plain"))
+    refute @contract.valid?(response(@body + "\xFF".b))
+    refute @contract.valid?(response(@body, "text/html", "500"))
+  end
+
+  def test_intermittent_200_error_page_rejects_the_loaded_run
+    server = TCPServer.new("127.0.0.1", 0)
+    worker = Thread.new do
+      loop do
+        socket = server.accept
+        count = 0
+        while socket.gets
+          while (line = socket.gets) && line != "\r\n"
+          end
+          body = count % 8 == 0 ? "Something went wrong" : @body
+          count += 1
+          socket.write("HTTP/1.1 200 OK\r\nContent-Type: text/html\r\nContent-Length: #{body.bytesize}\r\n\r\n#{body}")
+        end
+        socket.close
+      end
+    rescue IOError, SystemCallError
+      nil
+    end
+    client = BenchmarkHTTPClient.new("http://127.0.0.1:#{server.addr[1]}")
+    error = assert_raises(RuntimeError) { client.measure("/rooms/1", "", concurrency: 1, duration: 0.1, contract: @contract) }
+    assert_match(/HTTP statuses \{"200"\s*=>\s*\d+\}, 0 transport errors, [1-9]\d* invalid bodies/, error.message)
+  ensure
+    server&.close
+    worker&.kill
+    worker&.join
+  end
+
+  private
+    def response(body, type = "text/html", status = "200")
+      Net::HTTPResponse.new("1.1", status, "OK").tap do |result|
+        result["content-type"] = type
+        result.instance_variable_set(:@body, body)
+        result.instance_variable_set(:@read, true)
+      end
+    end
+end
@@ -80,5 +80,5 @@ Rails.application.configure do
  config.action_controller.raise_on_missing_callback_actions = true

  # Visit /rails/locks to see the locks
-  config.middleware.insert_before Rack::Sendfile, ActionDispatch::DebugLocks
+  config.middleware.insert_before ActionDispatch::Executor, ActionDispatch::DebugLocks
 end
@@ -1,5 +1,20 @@
+require "rails_ext/time_limited_video_previewer"
+
 ActiveSupport.on_load(:active_storage_blob) do
  ActiveStorage::DiskController.after_action only: :show do
    response.set_header("Cache-Control", "max-age=3600, public")
  end
 end
+
+Rails.application.configure do
+  # Rails' filter takes the second of the frames it selects (the first one, keyframes, scene changes), so a video
+  # with a single keyframe and no scene change is decoded to its end. Selecting any frame from 5 seconds on stops
+  # it there.
+  config.active_storage.video_preview_arguments =
+    "-vf 'select=eq(n\\,0)+eq(key\\,1)+gt(scene\\,0.015)+gte(t\\,5),loop=loop=-1:size=2,trim=start_frame=1'" \
+    " -frames:v 1 -f image2"
+
+  config.active_storage.previewers = config.active_storage.previewers.map do |previewer|
+    previewer == ActiveStorage::Previewer::VideoPreviewer ? TimeLimitedVideoPreviewer : previewer
+  end
+end
@@ -6,3 +6,11 @@ if Rails.env.production? && ENV["SKIP_TELEMETRY"].blank?
    config.release = ENV["GIT_REVISION"]
  end
 end
+
+# sentry-rails declares its Action Cable handle_open and handle_close wrappers private,
+# but Rails 8.2 calls them from outside the connection, so every /cable connection fails.
+# Remove once sentry-rails ships getsentry/sentry-ruby#2972 (issue #2975).
+ActiveSupport.on_load(:action_cable_connection) do
+  require "sentry/rails/action_cable"
+  Sentry::Rails::ActionCableExtensions::Connection.send(:public, :handle_open, :handle_close)
+end
@@ -1,2 +1,2 @@
 # Used to match JavaScripts (new Date).getTime() for sorting
-Time::DATE_FORMATS[:epoch] = ->(time) { (time.to_f * 1000).to_i }
+ActiveSupport::TimeFormats.register(:epoch, ->(time) { (time.to_f * 1000).to_i })
@@ -1,4 +1,5 @@
 require "web-push"
+require "web_push/connections"
 require "web_push/pool"
 require "web_push/notification"

@@ -17,43 +18,31 @@ end

 module WebPush::PersistentRequest
  def perform
-    if endpoint_ip = @options[:endpoint_ip]
-      # Pin the connection to the public IP resolved (and guarded) by
-      # Push::Subscription so delivery can't be rebound to a private address
-      # between resolution and connect. Bypasses the shared persistent pool,
-      # which would re-resolve the host itself.
-      #
-      # The explicit nil proxy address disables proxy discovery from
-      # http_proxy/https_proxy. An egress proxy would open the TCP connection
-      # itself and re-resolve the endpoint host, so http.ipaddr would no longer
-      # pin the destination and the DNS-rebinding guarantee would be lost. This
-      # path is already committed to a direct connection (it bypasses the pool);
-      # push delivery to public vendor endpoints goes direct.
-      http = Net::HTTP.new(uri.host, uri.port, nil)
-      http.ipaddr = endpoint_ip
-      http.use_ssl = true
-      http.ssl_timeout = @options[:ssl_timeout] unless @options[:ssl_timeout].nil?
-      http.open_timeout = @options[:open_timeout] unless @options[:open_timeout].nil?
-      http.read_timeout = @options[:read_timeout] unless @options[:read_timeout].nil?
-    elsif @options[:connection]
-      http = @options[:connection]
-    else
-      http = Net::HTTP.new(uri.host, uri.port, *proxy_options)
-      http.use_ssl = true
-      http.ssl_timeout = @options[:ssl_timeout] unless @options[:ssl_timeout].nil?
-      http.open_timeout = @options[:open_timeout] unless @options[:open_timeout].nil?
-      http.read_timeout = @options[:read_timeout] unless @options[:read_timeout].nil?
-    end
+    # Pin the connection to the public IP resolved (and guarded) by
+    # Push::Subscription so delivery can't be rebound to a private address
+    # between resolution and connect. There is no unpinned path: a delivery
+    # without a resolved IP is not sent.
+    endpoint_ip = @options[:endpoint_ip] or raise ArgumentError, "Push deliveries must be pinned to a resolved endpoint IP"
+
+    # The explicit nil proxy address disables proxy discovery from
+    # http_proxy/https_proxy. An egress proxy would open the TCP connection
+    # itself and re-resolve the endpoint host, so http.ipaddr would no longer
+    # pin the destination and the DNS-rebinding guarantee would be lost.
+    # Push delivery to public vendor endpoints goes direct.
+    http = Net::HTTP.new(uri.host, uri.port, nil)
+    http.ipaddr = endpoint_ip
+    http.use_ssl = true
+    http.ssl_timeout = @options[:ssl_timeout] unless @options[:ssl_timeout].nil?
+    http.open_timeout = @options[:open_timeout] unless @options[:open_timeout].nil?
+    http.read_timeout = @options[:read_timeout] unless @options[:read_timeout].nil?

    req = Net::HTTP::Post.new(uri.request_uri, headers)
    req.body = body

-    if http.is_a?(Net::HTTP::Persistent)
-      response = http.request uri, req
-    else
-      resp = http.request(req)
-      verify_response(resp)
-    end
+    # WebPush::Connections reuses an open connection only for this same host
+    # and pinned IP, so the guarantee holds for every request it sends.
+    resp = @options[:connection] ? @options[:connection].request(http, req) : http.request(req)
+    verify_response(resp)

    resp
  end
@@ -1,5 +1,5 @@
 class AddRoomIdAndCreatedAtIndexToMessages < ActiveRecord::Migration[8.2]
  def change
-    add_index :messages, %i[ room_id created_at ]
+    add_index :messages, %i[ room_id created_at ], if_not_exists: true
  end
 end
@@ -0,0 +1,5 @@
+class AddRoomIdAndUpdatedAtIndexToMessages < ActiveRecord::Migration[8.2]
+  def change
+    add_index :messages, %i[ room_id updated_at ], if_not_exists: true
+  end
+end
@@ -10,7 +10,7 @@
 #
 # It's strongly recommended that you check this file into your version control system.

-ActiveRecord::Schema[8.2].define(version: 2026_10_04_225000) do
+ActiveRecord::Schema[8.2].define(version: 2026_10_05_022000) do
  create_table "accounts", force: :cascade do |t|
    t.datetime "created_at", null: false
    t.text "custom_styles"
@@ -102,6 +102,7 @@ ActiveRecord::Schema[8.2].define(version: 2026_10_04_225000) do
    t.datetime "updated_at", null: false
    t.index ["creator_id"], name: "index_messages_on_creator_id"
    t.index ["room_id", "created_at"], name: "index_messages_on_room_id_and_created_at"
+    t.index ["room_id", "updated_at"], name: "index_messages_on_room_id_and_updated_at"
    t.index ["room_id"], name: "index_messages_on_room_id"
  end

@@ -0,0 +1,27 @@
+# Rails waits for ffmpeg however long it takes, and a video's preview is drawn inside the request that posts it.
+# This previewer gives ffmpeg a wall-clock limit, kills it past that, and reports a failed preview, so the message
+# is posted without one.
+class TimeLimitedVideoPreviewer < ActiveStorage::Previewer::VideoPreviewer
+  TIME_LIMIT = 10 # seconds
+
+  private
+    def capture(*argv, to:)
+      to.binmode
+
+      open_tempfile do |err|
+        IO.popen(argv, in: IO::NULL, err: err) do |out|
+          Timeout.timeout(TIME_LIMIT) { IO.copy_stream(out, to) }
+        rescue Timeout::Error
+          Process.kill :KILL, out.pid
+          raise ActiveStorage::PreviewError, "#{argv.first} took longer than #{TIME_LIMIT} seconds"
+        end
+        err.rewind
+
+        unless $?.success?
+          raise ActiveStorage::PreviewError, "#{argv.first} failed (status #{$?.exitstatus}): #{err.read.to_s.chomp}"
+        end
+      end
+
+      to.rewind
+    end
+end
@@ -0,0 +1,120 @@
+# This is in lib so we can use it in a thread pool without the Rails executor
+#
+# Keeps TLS connections to push services open between deliveries, so a push doesn't pay for a new TCP and TLS
+# handshake every time. Each connection is pinned to the address Push::Subscription resolved and guarded for the
+# delivery that opened it, and is handed out again only to a delivery whose own resolution, just made, returned that
+# same address for the same host. Net::HTTP reconnects to its pinned address, never to a new lookup of the host.
+class WebPush::Connections
+  class ConnectionLost < StandardError; end
+  class StaleConnection < StandardError; end
+
+  # Where a request failed. Before writing it, Net::HTTP checks an idle connection (:checking) and connects again
+  # if the push service closed it (:connecting); then it writes (:sent). Only a failed check means a dead idle
+  # connection that a new one can replace without sending the push twice.
+  module Stages
+    attr_reader :stage
+
+    private
+      def begin_transport(...)
+        @stage = :checking
+        super.tap { @stage = :sent }
+      end
+
+      def connect(...)
+        @stage = :connecting if @stage == :checking
+        super
+      end
+  end
+
+  def initialize(keep_alive_timeout: 30, max_idle: 150)
+    @keep_alive_timeout = keep_alive_timeout
+    @max_idle = max_idle
+    @idle = Hash.new { |idle, address| idle[address] = [] }
+    @mutex = Mutex.new
+    @pid = Process.pid
+  end
+
+  def request(http, request)
+    unless http.ipaddr && !http.proxy? && http.use_ssl? && !http.started?
+      raise ArgumentError, "Only new, direct TLS connections pinned to an address are pooled"
+    end
+    address = [ http.address, http.port, http.ipaddr ]
+
+    if idle = checkout(address)
+      begin
+        return send_over(idle, request, reused: true).tap { checkin(address, idle) }
+      rescue StaleConnection
+        # The push service had closed it: a new connection takes the push
+      end
+    end
+
+    http.extend Stages
+    http.keep_alive_timeout = @keep_alive_timeout
+    http.start
+    send_over(http, request, reused: false).tap { checkin(address, http) }
+  end
+
+  def shutdown
+    @mutex.synchronize do
+      forget_after_fork
+      @shut_down = true
+      @idle.each_value { |connections| connections.each { |http, _| close(http) } }
+      @idle.clear
+    end
+  end
+
+  private
+    def send_over(http, request, reused:)
+      http.request(request)
+    rescue IOError, SystemCallError, OpenSSL::SSL::SSLError => error
+      close(http)
+      # Only a connection that was idle can have been dead already: on a new one the error is the push service's
+      raise StaleConnection if reused && http.stage == :checking
+      # The push service may have it: don't send it twice, and don't report a dropped connection as a TLS failure
+      raise ConnectionLost, "#{error.class}: #{error.message}" if http.stage == :sent
+      raise
+    end
+
+    def checkout(address)
+      @mutex.synchronize do
+        forget_after_fork
+        close_expired
+        @idle[address].pop&.first
+      end
+    end
+
+    def checkin(address, http)
+      @mutex.synchronize do
+        close_expired
+        if @shut_down || @idle.values.sum(&:size) >= @max_idle
+          close(http)
+        else
+          @idle[address].push [ http, now ]
+        end
+      end
+    end
+
+    # A forked process must not write on its parent's TLS sessions. They're left to the parent, not closed.
+    def forget_after_fork
+      unless @pid == Process.pid
+        @idle = Hash.new { |idle, address| idle[address] = [] }
+        @pid = Process.pid
+      end
+    end
+
+    def close_expired
+      @idle.each_value do |connections|
+        connections.reject! { |http, idle_since| (now - idle_since > @keep_alive_timeout).tap { |expired| close(http) if expired } }
+      end
+      @idle.delete_if { |_, connections| connections.empty? }
+    end
+
+    def close(http)
+      http.finish if http.started?
+    rescue IOError, SystemCallError, OpenSSL::SSL::SSLError
+    end
+
+    def now
+      Process.clock_gettime(Process::CLOCK_MONOTONIC)
+    end
+end
@@ -5,13 +5,17 @@ class WebPush::Pool
  def initialize(invalid_subscription_handler:)
    @delivery_pool = Concurrent::ThreadPoolExecutor.new(max_threads: 50, max_queue: 10000)
    @invalidation_pool = Concurrent::FixedThreadPool.new(1)
-    @connection = Net::HTTP::Persistent.new(name: "web_push", pool_size: 150)
+    @connection = WebPush::Connections.new
    @invalid_subscription_handler = invalid_subscription_handler
  end

  def queue(payload, subscriptions)
-    subscriptions.find_each do |subscription|
-      deliver_later(payload, subscription)
+    subscriptions.find_in_batches do |batch|
+      unread_counts = Membership.unread.where(user_id: batch.map(&:user_id)).group(:user_id).count
+
+      batch.each do |subscription|
+        deliver_later(payload, subscription, badge: unread_counts.fetch(subscription.user_id, 0))
+      end
    end
  end

@@ -22,9 +26,9 @@ class WebPush::Pool
  end

  private
-    def deliver_later(payload, subscription)
+    def deliver_later(payload, subscription, badge:)
      # Ensure any AR operations happen before we post to the thread pool
-      notification = subscription.notification(**payload)
+      notification = subscription.notification(**payload, badge: badge)
      subscription_id = subscription.id

      delivery_pool.post do
@@ -40,6 +40,16 @@ class PresenceChannelTest < ActionCable::Channel::TestCase
    end
  end

+  test "subscribing tells the user's other tabs when they read the room" do
+    membership = users(:david).memberships.first
+
+    freeze_time do
+      assert_broadcast_on "user_#{users(:david).id}_reads", { room_id: membership.room_id, at: Time.current.to_f } do
+        subscribe room_id: membership.room_id
+      end
+    end
+  end
+
  test "unsubscribing marks the membership as disconnected" do
    membership = users(:david).memberships.first
    subscribe room_id: membership.room_id
@@ -8,7 +8,7 @@ class UnreadRoomsChannelTest < ActionCable::Channel::TestCase

    assert subscription.confirmed?
    assert_has_stream "user_#{users(:jz).id}_unreads"
-    assert_not_includes subscription.streams, "unread_rooms"
+    assert_not_includes subscription.stream_names, "unread_rooms"
  end

  test "an outsider is not told about activity in a room they can't see" do
@@ -16,7 +16,9 @@ class UnreadRoomsChannelTest < ActionCable::Channel::TestCase
    assert_not direct.users.include?(users(:jz)), "jz must be an outsider for this test to mean anything"

    broadcasts = capture_unread_broadcasts_for(users(:jz)) do
-      direct.messages.create!(body: "Private", creator: users(:kevin), client_message_id: "outsider").broadcast_create
+      perform_enqueued_jobs only: Message::BroadcastUnreadRoomJob do
+        direct.messages.create!(body: "Private", creator: users(:kevin), client_message_id: "outsider").broadcast_create
+      end
    end

    assert_empty broadcasts
@@ -24,12 +26,16 @@ class UnreadRoomsChannelTest < ActionCable::Channel::TestCase

  test "a member is told about activity in their own room" do
    direct = rooms(:bender_and_kevin)
+    message = nil

    broadcasts = capture_unread_broadcasts_for(users(:kevin)) do
-      direct.messages.create!(body: "Private", creator: users(:bender), client_message_id: "member").broadcast_create
+      perform_enqueued_jobs only: Message::BroadcastUnreadRoomJob do
+        message = direct.messages.create!(body: "Private", creator: users(:bender), client_message_id: "member")
+        message.broadcast_create
+      end
    end

-    assert_equal [ direct.id ], broadcasts.collect { |broadcast| broadcast["roomId"] }
+    assert_equal [ { "roomId" => direct.id, "at" => message.created_at.to_f } ], broadcasts
  end

  private
@@ -37,7 +43,7 @@ class UnreadRoomsChannelTest < ActionCable::Channel::TestCase
      stub_connection(current_user: user)
      subscribe

-      stream = subscription.streams.sole
+      stream = subscription.stream_names.sole
      before = ActionCable.server.pubsub.broadcasts(stream).size

      yield
@@ -10,6 +10,16 @@ class AccountsControllerTest < ActionDispatch::IntegrationTest
    assert_response :ok
  end

+  test "edit shows whether room creation is restricted to administrators" do
+    get edit_account_url
+    assert_select "label.switch input.switch__input[type=checkbox]:not([checked])"
+
+    accounts(:signal).update!(settings: { restrict_room_creation_to_administrators: true })
+
+    get edit_account_url
+    assert_select "label.switch input.switch__input[type=checkbox][checked][data-action='change->form#submit']"
+  end
+
  test "edit groups administrators separately from members with a divider" do
    get edit_account_url

@@ -42,6 +52,19 @@ class AccountsControllerTest < ActionDispatch::IntegrationTest
    end
  end

+  test "edit lists a page of members and the next page carries on from it" do
+    User.insert_all 501.times.map { |i| { name: "Member #{i}", email_address: "member#{i}@37signals.com" } }
+    users(:kevin).banned!
+
+    get edit_account_url
+    first_page = listed_user_ids
+    get account_users_url(page: 2, format: :turbo_stream)
+    next_page = listed_user_ids
+
+    assert_not_empty next_page
+    assert_equal User.where(status: [ :active, :banned ]).without_bots.ids.sort, (first_page + next_page).sort
+  end
+
  test "update" do
    assert users(:david).administrator?

@@ -58,4 +81,9 @@ class AccountsControllerTest < ActionDispatch::IntegrationTest
    put account_url, params: { account: { name: "Different" } }
    assert_response :forbidden
  end
+
+  private
+    def listed_user_ids
+      response.body.scan(%r{href="/users/(\d+)"}).flatten.map(&:to_i)
+    end
 end
@@ -13,6 +13,15 @@ class Messages::BoostsControllerTest < ActionDispatch::IntegrationTest
    assert_select ".message__boost-inline a.boost__action[data-action='soft-keyboard#open']"
  end

+  test "index looks up the boosters of all boosts at once" do
+    @message.boosts.create! booster: users(:jason), content: "🎉"
+    queries_with_two_boosts = count_queries { get message_boosts_url(@message) }
+
+    boost = @message.boosts.create! booster: users(:kevin), content: "👀"
+    assert_equal queries_with_two_boosts, count_queries { get message_boosts_url(@message) }
+    assert_select "#" + dom_id(boost)
+  end
+
  test "create" do
    assert_turbo_stream_broadcasts [ @message.room, :messages ], count: 1 do
      assert_difference -> { @message.boosts.count }, 1 do
@@ -30,4 +39,12 @@ class Messages::BoostsControllerTest < ActionDispatch::IntegrationTest
      end
    end
  end
+
+  private
+    def count_queries(&block)
+      count = 0
+      counter = ->(*, payload) { count += 1 unless payload[:name] == "SCHEMA" || payload[:cached] }
+      ActiveSupport::Notifications.subscribed(counter, "sql.active_record", &block)
+      count
+    end
 end
@@ -27,6 +27,62 @@ class MessagesCachingTest < ActionDispatch::IntegrationTest
    end
  end

+  test "boosts are cached inside their message instead of one fragment each" do
+    with_memory_cache do
+      cache_keys = []
+      subscriber = ActiveSupport::Notifications.subscribe(/\Acache_(read|read_multi|write|write_multi)\.active_support\z/) do |*, payload|
+        cache_keys.concat(payload[:key].is_a?(Hash) ? payload[:key].keys : Array(payload[:key]))
+      end
+
+      get room_messages_url(rooms(:watercooler))
+      assert_response :success
+      assert_select "#" + dom_id(boosts(:fourth_by_bender))
+      assert_select "#" + dom_id(boosts(:thirteenth))
+      assert_empty cache_keys.grep(%r{messages/boosts/_boost})
+
+      boost = messages(:fourth).boosts.create! booster: users(:jason), content: "🎉"
+      get room_messages_url(rooms(:watercooler))
+      refreshed = response.body
+
+      Rails.cache.clear
+      get room_messages_url(rooms(:watercooler))
+      assert_equal refreshed, response.body
+      assert_select "##{dom_id(messages(:fourth))} ##{dom_id(boost)}", text: /🎉/
+    ensure
+      ActiveSupport::Notifications.unsubscribe(subscriber)
+    end
+  end
+
+  test "messages render their preloaded boosts in order without querying them again" do
+    earlier = messages(:fourth).boosts.create! booster: users(:jason), content: "🥇", created_at: 1.day.ago
+    in_order = [ dom_id(earlier), dom_id(boosts(:fourth_by_bender)) ]
+
+    assert_no_queries_match(/ORDER BY "boosts"/) do
+      get room_messages_url(rooms(:watercooler))
+    end
+    assert_response :success
+    assert_equal in_order, css_select("##{dom_id(messages(:fourth), :boosts)} .boost").map { it["id"] }
+
+    get message_boosts_url(messages(:fourth))
+    assert_equal in_order, css_select("##{dom_id(messages(:fourth), :boosts)} .boost").map { it["id"] }
+  end
+
+  test "a page of messages loads whether their previews were made along with the messages, not one at a time" do
+    room = rooms(:watercooler)
+    2.times do |copy|
+      { "moon.jpg" => "image/jpeg", "alpha-centuri.mov" => "video/quicktime" }.each do |file, content_type|
+        room.messages.create_with_attachment! creator: users(:david), client_message_id: "#{copy}-#{file}", attachment: fixture_file_upload(file, content_type)
+      end
+    end
+
+    assert_no_queries_match(/"active_storage_\w+"\."(id|blob_id|record_id)" = \?/) do
+      get room_messages_url(room)
+    end
+    assert_response :success
+    assert_select "img[src*='moon.jpg']", count: 2
+    assert_select "video[poster]", count: 2
+  end
+
  private
    def with_memory_cache
      old_cache = Rails.cache
@@ -57,21 +57,89 @@ class MessagesControllerTest < ActionDispatch::IntegrationTest
    end
  end

+  test "message copy links preserve the request origin in broadcasts and regular reads" do
+    [
+      [ "once.campfire.test:38683", false, "http://once.campfire.test:38683" ],
+      [ "once.campfire.test:8443", true, "https://once.campfire.test:8443" ],
+      [ "once.campfire.test:80", false, "http://once.campfire.test" ],
+      [ "once.campfire.test:443", true, "https://once.campfire.test" ]
+    ].each_with_index do |(host, secure, origin), index|
+      host! host
+      https! secure
+
+      post room_messages_path(@room, format: :turbo_stream), params: { message: {
+        body: "Request origin #{index}", client_message_id: "request-origin-#{index}" } }
+      assert_response :success
+      message = Message.last
+      expected_url = "#{origin}/rooms/#{@room.id}/@#{message.id}"
+
+      assert_rendered_turbo_stream_broadcast @room, :messages, action: "append", target: [ @room, :messages ] do
+        assert_copy_link_button expected_url
+      end
+
+      get room_message_path(@room, message)
+      assert_response :success
+      assert_copy_link_button expected_url
+    end
+  end
+
+  test "creating a message with an image that can't be decoded broadcasts the message to the room" do
+    webp = Vips::Image.new_from_file(file_fixture("moon.jpg").to_s).webpsave_buffer
+    broken = Rack::Test::UploadedFile.new(StringIO.new(webp.byteslice(0, webp.bytesize / 2)), "image/webp", original_filename: "broken.webp")
+
+    post room_messages_url(@room, format: :turbo_stream), params: { message: { attachment: broken, client_message_id: 999 } }
+
+    assert_response :success
+    assert_rendered_turbo_stream_broadcast @room, :messages, action: "append", target: [ @room, :messages ] do
+      assert_select ".message__body a[href*='broken.webp']"
+    end
+  end
+
  test "creating a message broadcasts unread room to each member" do
+    memberships(:david_watercooler).present # the poster is in the room
+
    @room.users.each do |member|
      assert_broadcasts UnreadRoomsChannel.stream_name_for(member.id), 1 do
-        post room_messages_url(@room, format: :turbo_stream), params: { message: { body: "New one #{member.id}", client_message_id: member.id } }
+        perform_enqueued_jobs only: Message::BroadcastUnreadRoomJob do
+          post room_messages_url(@room, format: :turbo_stream), params: { message: { body: "New one #{member.id}", client_message_id: member.id } }
+        end
      end
    end
  end

+  test "creating a message leaves the unread fanout to a job" do
+    member = @room.users.excluding(users(:david)).first
+
+    assert_no_broadcasts UnreadRoomsChannel.stream_name_for(member.id) do
+      post room_messages_url(@room, format: :turbo_stream), params: { message: { body: "New one", client_message_id: 999 } }
+    end
+
+    assert_enqueued_with job: Message::BroadcastUnreadRoomJob, args: [ Message.last ]
+  end
+
+  test "the unread fanout skips a member who has opened the room since" do
+    membership = memberships(:jason_watercooler)
+
+    post room_messages_url(@room, format: :turbo_stream), params: { message: { body: "New one", client_message_id: 999 } }
+    assert membership.reload.unread?
+
+    membership.present
+    membership.reload.disconnected
+
+    assert_no_broadcasts UnreadRoomsChannel.stream_name_for(membership.user_id) do
+      perform_enqueued_jobs only: Message::BroadcastUnreadRoomJob
+    end
+  end
+
  test "creating a message doesn't broadcast unread room to non-members" do
    outsiders = User.where.not(id: @room.users.map(&:id))
    assert outsiders.any?, "need someone outside the room for this test to mean anything"

    outsiders.each do |outsider|
      assert_no_broadcasts UnreadRoomsChannel.stream_name_for(outsider.id) do
-        post room_messages_url(@room, format: :turbo_stream), params: { message: { body: "New one", client_message_id: 999 } }
+        perform_enqueued_jobs only: Message::BroadcastUnreadRoomJob do
+          post room_messages_url(@room, format: :turbo_stream), params: { message: { body: "New one", client_message_id: 999 } }
+        end
      end
    end
  end
@@ -148,6 +216,20 @@ class MessagesControllerTest < ActionDispatch::IntegrationTest
    end
  end

+  test "messages sharing a client_message_id render as distinct elements" do
+    victim = @messages.last
+
+    post room_messages_url(@room, format: :turbo_stream), params: { message: { body: "Collision", client_message_id: victim.client_message_id } }
+    colliding = @room.messages.ordered.last
+    assert_not_equal victim, colliding
+
+    get room_messages_url(@room)
+
+    ids = css_select(".message[data-message-id]").map { it["id"] }
+    assert_equal ids.uniq, ids
+    ensure_messages_present victim, colliding
+  end
+
  private
    def ensure_messages_present(*messages, count: 1)
      messages.each do |message|
@@ -25,8 +25,10 @@ class Rooms::DirectsControllerTest < ActionDispatch::IntegrationTest
    sign_in :kevin

    assert_difference -> { Room.count }, -1 do
-      delete rooms_direct_url(rooms(:david_and_kevin))
-      assert_redirected_to root_url
+      perform_enqueued_jobs do
+        delete rooms_direct_url(rooms(:david_and_kevin))
+        assert_redirected_to root_url
+      end
    end
  end

@@ -15,6 +15,21 @@ class RoomsControllerTest < ActionDispatch::IntegrationTest
    assert_response :success
  end

+  test "show renders notification help for each platform" do
+    {
+      "Firefox on Android" => "Mozilla/5.0 (Android 14; Mobile; rv:131.0) Gecko/131.0 Firefox/131.0",
+      "Chrome on Android" => "Mozilla/5.0 (Linux; Android 14; Pixel 8) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/141.0.0.0 Mobile Safari/537.36",
+      "Firefox on desktop" => "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:131.0) Gecko/20100101 Firefox/131.0",
+      "Chrome on desktop" => "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/141.0.0.0 Safari/537.36",
+      "Safari on macOS" => "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/18.0 Safari/605.1.15"
+    }.each do |platform, user_agent|
+      get room_url(users(:david).rooms.last), headers: { "User-Agent" => user_agent }
+
+      assert_response :success, platform
+      assert_select ".notifications-help ol", { minimum: 1 }, platform
+    end
+  end
+
  test "shows records the last room visited in a cookie" do
    get room_url(users(:david).rooms.last)
    assert response.cookies[:last_room] = users(:david).rooms.last.id
@@ -68,11 +83,24 @@ class RoomsControllerTest < ActionDispatch::IntegrationTest
  test "destroy" do
    assert_turbo_stream_broadcasts :rooms, count: 1 do
      assert_difference -> { Room.count }, -1 do
-        delete room_url(rooms(:designers))
+        perform_enqueued_jobs { delete room_url(rooms(:designers)) }
      end
    end
  end

+  test "destroy takes the room away from its members at once and leaves its messages to a job" do
+    room = rooms(:designers)
+    member_ids = room.memberships.pluck(:user_id)
+
+    assert_enqueued_with(job: Room::DestroyJob, args: [ room, member_ids ]) do
+      assert_no_difference -> { Message.count } do
+        delete room_url(room)
+        assert_redirected_to root_url
+      end
+    end
+    assert_empty room.memberships.reload
+  end
+
  test "destroy only allowed for creators or those who can administer" do
    sign_in :jz

@@ -84,7 +112,7 @@ class RoomsControllerTest < ActionDispatch::IntegrationTest
    rooms(:designers).update! creator: users(:jz)

    assert_difference -> { Room.count }, -1 do
-      delete room_url(rooms(:designers))
+      perform_enqueued_jobs { delete room_url(rooms(:designers)) }
    end
  end

@@ -5,6 +5,10 @@ class Sessions::TransfersControllerTest < ActionDispatch::IntegrationTest
    get session_transfer_url("some-token")

    assert_response :success
+    assert_select "form[data-controller='auto-submit']", count: 1
+    assert_select "input[name='_method'][value='put']", count: 1
+    assert_equal response.body.scan(/<form\b/).size, response.body.scan("</form>").size
+    assert_match(/<form\b[^>]*data-controller="auto-submit"[^>]*>(?:\s*<input\b[^>]*>)*\s*<\/form>/, response.body)
  end

  test "update establishes a session when the code is valid" do
@@ -75,7 +75,94 @@ class UnfurlLinksControllerTest < ActionDispatch::IntegrationTest
    assert_equal "Hey!", JSON.parse(response.body)["title"]
  end

+  test "create gives up on a host that accepts the connection and never answers" do
+    with_local_host do |url, connections|
+      stub_const(Opengraph::Fetch, :TIMEOUT, 0.2.seconds) do
+        assert_unfurls_nothing_within(1.second) { post unfurl_link_url, params: { url: url } }
+      end
+      assert_equal 1, connections.size, "the GET is not retried"
+    end
+  end
+
+  test "create gives up on headers that trickle in past the deadline" do
+    trickle_headers = ->(client) do
+      client.write "HTTP/1.1 200 OK\r\nX-Padding: "
+      100.times { client.write "x"; sleep 0.05 }
+      client.write "\r\nContent-Type: text/html\r\nContent-Length: 2\r\n\r\nok"
+    end
+
+    with_local_host(trickle_headers) do |url|
+      stub_const(UnfurlLinksController, :DEADLINE, 0.5.seconds) do
+        assert_unfurls_nothing_within(1.5.seconds) { post unfurl_link_url, params: { url: url } }
+      end
+    end
+  end
+
+  test "create gives up on a body that trickles in past the deadline" do
+    trickle_body = ->(client) do
+      client.write "HTTP/1.1 200 OK\r\nContent-Type: text/html\r\nContent-Length: 100\r\n\r\n"
+      100.times { client.write "x"; sleep 0.05 }
+    end
+
+    with_local_host(trickle_body) do |url|
+      stub_const(UnfurlLinksController, :DEADLINE, 0.5.seconds) do
+        assert_unfurls_nothing_within(1.5.seconds) { post unfurl_link_url, params: { url: url } }
+      end
+    end
+  end
+
+  test "create gives up on redirects that keep coming past the deadline" do
+    stub_dns_resolution("1.2.3.4")
+    WebMock.stub_request(:get, "https://www.example.com/").to_return do
+      sleep 0.2
+      { status: 302, headers: { location: "https://www.example.com/" } }
+    end
+
+    stub_const(UnfurlLinksController, :DEADLINE, 0.5.seconds) do
+      assert_unfurls_nothing_within(1.second) { post unfurl_link_url, params: { url: "https://www.example.com" } }
+    end
+  end
+
+  test "create gives up on a host name that takes too long to look up" do
+    Resolv.stubs(:getaddresses).with { sleep 2 }.returns([ "1.2.3.4" ]) # each lookup takes 2 s
+    WebMock.stub_request(:get, "https://www.example.com/").to_return(status: 200, body: "<html></html>", headers: { content_type: "text/html" })
+
+    stub_const(UnfurlLinksController, :DEADLINE, 0.5.seconds) do
+      assert_unfurls_nothing_within(1.second) { post unfurl_link_url, params: { url: "https://www.example.com" } }
+    end
+  end
+
  private
+    # A real socket, because WebMock reads the whole response before the code under test sees any of it.
+    # The guard lets the local address through, as it would a public one.
+    def with_local_host(respond = ->(client) { })
+      server = TCPServer.new("127.0.0.1", 0)
+      connections = Queue.new
+      Thread.new do
+        loop do
+          client = server.accept
+          connections << client
+          client.readpartial(1024)
+          respond.call(client)
+        end
+      rescue IOError, SystemCallError
+      end
+
+      RestrictedHTTP::PrivateNetworkGuard.stubs(:resolve).returns("127.0.0.1")
+      WebMock.disable!
+      yield "http://www.example.com:#{server.addr[1]}/", connections
+    ensure
+      WebMock.enable!
+      server&.close
+    end
+
+    def assert_unfurls_nothing_within(limit)
+      started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+      yield
+      assert_response :no_content
+      assert_operator Process.clock_gettime(Process::CLOCK_MONOTONIC) - started, :<, limit
+    end
+
    def stub_successful_request(url: "https://www.example.com/", title: "Hey!", description: "desc..")
      WebMock.stub_request(:get, url).to_return(
        status: 200,
@@ -11,6 +11,13 @@ class Users::ProfilesControllerTest < ActionDispatch::IntegrationTest
    assert_response :success
  end

+  test "show includes install instructions for Edge" do
+    get user_profile_url, headers: { "User-Agent" => "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/70.0.3538.102 Safari/537.36 Edge/18.18363" }
+
+    assert_response :success
+    assert_select ".pwa__instructions img[src*='install-edge']"
+  end
+
  test "update" do
    put user_profile_url, params: { user: { name: "John Doe", bio: "Acrobat" } }

@@ -27,4 +27,25 @@ class Users::SidebarsControllerTest < ActionDispatch::IntegrationTest
    get user_sidebar_url
    assert_select ".unread", count: users(:david).memberships.reject { |m| m.room.direct? || !m.unread? }.count
  end
+
+  test "directs are ordered by room recency, not name" do
+    older = rooms(:david_and_jason)
+    newer = rooms(:david_and_kevin)
+    older.update_column :updated_at, 2.days.ago
+    newer.update_column :updated_at, 1.minute.ago
+
+    get user_sidebar_url
+
+    assert_operator @response.body.index(dom_id(newer, :list)), :<, @response.body.index(dom_id(older, :list))
+  end
+
+  test "shared rooms stay ordered by name" do
+    get user_sidebar_url
+
+    shared_ids = users(:david).rooms.without_directs.sort_by { |room| room.name.downcase }.map { |room| dom_id(room, :list) }
+    positions = shared_ids.map { |id| @response.body.index(id) }
+
+    assert positions.all?
+    assert_equal positions.sort, positions
+  end
 end
@@ -1,4 +1,4 @@
-<% password_digest = BCrypt::Password.create("secret123456") %>
+<% password_digest = BCrypt::Password.create("secret123456", cost: BCrypt::Engine::MIN_COST) %>

 david:
  name: David
@@ -24,4 +24,86 @@ class MessagesHelperTest < ActionView::TestCase
    assert_match /<a href="https:\/\/example\.com"[^>]*>example<\/a>/, presentation
    assert_match /<strong>bold<\/strong>/, presentation
  end
+
+  test "message_presentation shows an image's thumbnail made when it was posted" do
+    presentation = view.message_presentation(attachment_message("moon.jpg", "image/jpeg", processed: true))
+
+    assert_match %r{<img[^>]+src="[^"]*/representations/[^"]*moon\.jpg"}, presentation
+  end
+
+  test "message_presentation links an image whose thumbnail wasn't made, rather than making it on view" do
+    presentation = view.message_presentation(attachment_message("moon.jpg", "image/jpeg", processed: false))
+
+    assert_no_match %r{/representations/}, presentation
+    assert_match %r{<span>moon\.jpg</span>}, presentation
+  end
+
+  test "message_presentation gives a video the poster made when it was posted" do
+    presentation = view.message_presentation(attachment_message("alpha-centuri.mov", "video/quicktime", processed: true))
+
+    assert_match %r{<video[^>]+poster="[^"]*/representations/[^"]*alpha-centuri}, presentation
+  end
+
+  test "message_presentation shows a video whose poster wasn't made without one, rather than making it on view" do
+    presentation = view.message_presentation(attachment_message("alpha-centuri.mov", "video/quicktime", processed: false))
+
+    assert_match %r{<video[^>]+src="[^"]*alpha-centuri\.mov"}, presentation
+    assert_no_match %r{poster=|/representations/}, presentation
+  end
+
+  test "message_presentation shows a video whose frame was drawn but whose poster wasn't made without one" do
+    message = attachment_message("alpha-centuri.mov", "video/quicktime", processed: false)
+    message.attachment.preview(format: :jpg).processed
+    assert message.attachment.preview(:poster).processed?
+
+    presentation = view.message_presentation(message.reload)
+
+    assert_match %r{<video[^>]+src="[^"]*alpha-centuri\.mov"}, presentation
+    assert_no_match %r{poster=|/representations/}, presentation
+  end
+
+  test "message_presentation shows an image in the rich text as a file, rather than making its preview on view" do
+    message = Message.with_attachment_details.find(message_with_file_in_rich_text("moon.jpg", "image/jpeg").id)
+
+    presentation = nil
+    assert_no_queries_match(/active_storage_variant_records/) { presentation = view.message_presentation(message) }
+
+    assert_no_match %r{/representations/}, presentation
+    assert_match %r{<span class="attachment__name">moon\.jpg</span>}, presentation
+  end
+
+  test "message_presentation shows an image in the rich text as a file even when its preview was made" do
+    message = message_with_file_in_rich_text("moon.jpg", "image/jpeg")
+    message.body.embeds.first.blob.variant(resize_to_limit: [ 1024, 768 ]).processed
+
+    presentation = view.message_presentation(message.reload)
+
+    assert_no_match %r{/representations/}, presentation
+    assert_match %r{<span class="attachment__name">moon\.jpg</span>}, presentation
+  end
+
+  test "message_presentation shows a video in the rich text as a file" do
+    presentation = view.message_presentation(message_with_file_in_rich_text("alpha-centuri.mov", "video/quicktime"))
+
+    assert_no_match %r{/representations/}, presentation
+    assert_match %r{<span class="attachment__name">alpha-centuri\.mov</span>}, presentation
+  end
+
+  private
+    def attachment_message(file, content_type, processed:)
+      attributes = { creator: users(:jason), client_message_id: "0015", attachment: fixture_file_upload(file, content_type) }
+
+      if processed
+        rooms(:pets).messages.create_with_attachment!(attributes)
+      else
+        rooms(:pets).messages.create!(attributes)
+      end
+    end
+
+    def message_with_file_in_rich_text(file, content_type)
+      blob = ActiveStorage::Blob.create_and_upload!(io: file_fixture(file).open, filename: file, content_type: content_type)
+      body = %(<div>Here: <action-text-attachment sgid="#{blob.attachable_sgid}"></action-text-attachment></div>)
+
+      Message.create! room: rooms(:pets), body: body, client_message_id: "0015", creator: users(:jason)
+    end
 end
@@ -0,0 +1,174 @@
+require "test_helper"
+
+class WebPush::ConnectionsTest < ActiveSupport::TestCase
+  include PushServiceTestHelper
+
+  setup { @connections = WebPush::Connections.new }
+  teardown { @connections.shutdown }
+
+  test "a delivery to the same host and address reuses the open connection" do
+    with_push_service do |server|
+      2.times { |i| assert_kind_of Net::HTTPCreated, @connections.request(pinned_connection(server), push_request("/push/#{i}")) }
+
+      assert_equal 1, server.connections
+      assert_equal %w[ /push/0 /push/1 ], server.requests
+    end
+  end
+
+  test "a delivery to another address never reuses a connection" do
+    with_push_service do |server|
+      @connections.request(pinned_connection(server), push_request)
+
+      other = pinned_connection(server, "127.0.0.2")
+      other.expects(:start)
+      other.expects(:request).returns(:sent_to_the_other_address)
+      assert_equal :sent_to_the_other_address, @connections.request(other, push_request)
+      assert_equal 1, server.requests.size
+    end
+  end
+
+  test "a connection the push service closed while idle is replaced, and the push sent once" do
+    [ :fin, :close_notify ].each do |hang_up|
+      with_push_service(hang_up_after_response: hang_up) do |server|
+        @connections.request(pinned_connection(server), push_request("/push/0"))
+        assert server.hung_up?
+
+        assert_kind_of Net::HTTPCreated, @connections.request(pinned_connection(server), push_request("/push/1"))
+        assert_equal 2, server.connections
+        assert_equal %w[ /push/0 /push/1 ], server.requests
+      end
+    end
+  end
+
+  test "a certificate for another name is refused, and nothing is sent" do
+    with_push_service(certificate: ->(_) { "other.test" }) do |server|
+      assert_raises(OpenSSL::SSL::SSLError) { @connections.request(pinned_connection(server), push_request) }
+      assert_empty server.requests
+    end
+  end
+
+  test "a certificate for another name is refused when a dead connection is replaced too" do
+    with_push_service(hang_up_after_response: :fin, certificate: ->(connection) { connection == 1 ? HOST : "other.test" }) do |server|
+      @connections.request(pinned_connection(server), push_request)
+      assert server.hung_up?
+
+      assert_raises(OpenSSL::SSL::SSLError) { @connections.request(pinned_connection(server), push_request) }
+      assert_equal 1, server.requests.size
+    end
+  end
+
+  test "a push the service may have received is not sent again" do
+    with_push_service(drop_request: ->(number) { number == 2 }) do |server|
+      @connections.request(pinned_connection(server), push_request("/push/0"))
+
+      error = assert_raises(WebPush::Connections::ConnectionLost) { @connections.request(pinned_connection(server), push_request("/push/1")) }
+      assert_not_kind_of OpenSSL::OpenSSLError, error
+      assert_equal %w[ /push/0 /push/1 ], server.requests
+      assert_equal 1, server.connections
+    end
+  end
+
+  test "a push the service may have received on a new connection isn't reported as a TLS failure either" do
+    with_push_service(drop_request: ->(number) { number == 1 }) do |server|
+      error = assert_raises(WebPush::Connections::ConnectionLost) { @connections.request(pinned_connection(server), push_request) }
+      assert_not_kind_of OpenSSL::OpenSSLError, error
+      assert_equal 1, server.requests.size
+    end
+  end
+
+  test "a TLS failure on a new connection before the push is written is reported as it is" do
+    failing_check = Module.new { private def begin_transport(*) = raise(OpenSSL::SSL::SSLError, "alert right after the handshake") }
+
+    with_push_service do |server|
+      assert_raises(OpenSSL::SSL::SSLError) { @connections.request(pinned_connection(server).extend(failing_check), push_request) }
+      assert_empty server.requests
+    end
+  end
+
+  test "a certificate for another name when reconnecting a cleanly closed connection is reported, even if a new one would work" do
+    certificates = { 2 => "other.test" }
+    with_push_service(hang_up_after_response: :close_notify, certificate: ->(connection) { certificates.fetch(connection, HOST) }) do |server|
+      @connections.request(pinned_connection(server), push_request)
+      assert server.hung_up?
+
+      assert_raises(OpenSSL::SSL::SSLError) { @connections.request(pinned_connection(server), push_request) }
+      assert_equal 1, server.requests.size
+      assert_equal 2, server.connections
+    end
+  end
+
+  test "only new, direct TLS connections pinned to an address are pooled" do
+    with_push_service do |server|
+      unpinned = Net::HTTP.new(HOST, server.port, nil).tap { it.use_ssl = true }
+      proxied = Net::HTTP.new(HOST, server.port, "127.0.0.1", 3128).tap { it.ipaddr = IP; it.use_ssl = true }
+      plain = pinned_connection(server).tap { it.use_ssl = false }
+      started = pinned_connection(server).tap(&:start)
+
+      [ unpinned, proxied, plain, started ].each do |http|
+        assert_raises(ArgumentError) { @connections.request(http, push_request) }
+      end
+      assert_empty server.requests
+    ensure
+      started&.finish
+    end
+  end
+
+  test "a forked process opens its own connections" do
+    with_push_service do |server|
+      @connections.request(pinned_connection(server), push_request)
+
+      child = Process.pid + 1
+      Process.stubs(:pid).returns(child)
+      @connections.request(pinned_connection(server), push_request)
+      assert_equal 2, server.connections
+    end
+  end
+
+  test "a forked process shutting down leaves its parent's connections open" do
+    with_push_service do |server|
+      @connections.request(pinned_connection(server), push_request)
+
+      child = Process.pid + 1
+      Process.stubs(:pid).returns(child)
+      @connections.shutdown
+      assert_not server.hung_up?(within: 0.5)
+    end
+  end
+
+  test "the idle pool closes connections beyond its bound" do
+    connections = WebPush::Connections.new(max_idle: 1)
+
+    with_push_service do |server|
+      other = Server.new
+      connections.request(pinned_connection(server), push_request)
+      connections.request(pinned_connection(other), push_request)
+      assert other.hung_up?
+      assert_not server.hung_up?(within: 0.1)
+
+      connections.request(pinned_connection(server), push_request)
+      assert_equal 1, server.connections
+    ensure
+      connections.shutdown
+      other&.stop
+    end
+  end
+
+  test "idle connections are closed after the keep-alive timeout and on shutdown" do
+    connections = WebPush::Connections.new(keep_alive_timeout: 0)
+
+    with_push_service do |server|
+      other = Server.new
+      connections.request(pinned_connection(server), push_request)
+      connections.request(pinned_connection(other), push_request)
+      assert server.hung_up?
+
+      connections.shutdown
+      assert other.hung_up?
+
+      connections.request(pinned_connection(server), push_request)
+      assert server.hung_up?
+    ensure
+      other&.stop
+    end
+  end
+end
@@ -1,6 +1,8 @@
 require "test_helper"

 class WebPush::PersistentRequestTest < ActiveSupport::TestCase
+  include PushServiceTestHelper
+
  ENDPOINT = "https://fcm.googleapis.com/fcm/send/test123"

  # The delivery must connect to the public IP resolved and guarded by
@@ -50,4 +52,72 @@ class WebPush::PersistentRequestTest < ActiveSupport::TestCase
    %w[ http_proxy https_proxy HTTP_PROXY HTTPS_PROXY ].each { |k| ENV.delete(k) }
    saved.each { |k, v| ENV[k] = v }
  end
+
+  test "deliveries through the pool to the same host and address share one connection" do
+    pool = WebPush::Pool.new(invalid_subscription_handler: nil)
+
+    with_push_service do |server|
+      2.times { deliver_to server, connection: pool.connection }
+
+      assert_equal 2, server.requests.size
+      assert_equal 1, server.connections
+    end
+  ensure
+    pool.shutdown
+  end
+
+  test "a pooled delivery checks the response like any other" do
+    pool = WebPush::Pool.new(invalid_subscription_handler: nil)
+
+    with_push_service(status: "410 Gone") do |server|
+      assert_raises(WebPush::ExpiredSubscription) { deliver_to server, connection: pool.connection }
+    end
+  ensure
+    pool.shutdown
+  end
+
+  test "pooled deliveries are pinned and ignore proxy env too" do
+    host = URI(ENDPOINT).host
+    saved = ENV.slice("http_proxy", "https_proxy", "HTTP_PROXY", "HTTPS_PROXY")
+    %w[ http_proxy https_proxy HTTP_PROXY HTTPS_PROXY ].each { |k| ENV[k] = "http://proxy.internal:3128" }
+
+    WebMock.disable_net_connect! allow: [ host ]
+
+    TCPSocket.expects(:open).with { |*args, **| args.first == host || args.first == "proxy.internal" }.never
+    TCPSocket.expects(:open).with { |*args, **| args.first == DnsTestHelper::WEB_PUSH_PUBLIC_TEST_IP && args[1] == 443 }.throws(:pinned_to_ip)
+
+    assert_throws :pinned_to_ip do
+      WebPush.payload_send \
+        message: "",
+        endpoint: ENDPOINT,
+        endpoint_ip: DnsTestHelper::WEB_PUSH_PUBLIC_TEST_IP,
+        p256dh: "", auth: "", vapid: {},
+        connection: WebPush::Connections.new,
+        urgency: "high"
+    end
+  ensure
+    %w[ http_proxy https_proxy HTTP_PROXY HTTPS_PROXY ].each { |k| ENV.delete(k) }
+    saved.each { |k, v| ENV[k] = v }
+  end
+
+  test "nothing is sent without a resolved endpoint IP" do
+    WebMock.disable_net_connect! allow: [ URI(ENDPOINT).host ]
+    TCPSocket.expects(:open).never
+
+    error = assert_raises(ArgumentError) do
+      WebPush.payload_send message: "", endpoint: ENDPOINT, p256dh: "", auth: "", vapid: {}, urgency: "high"
+    end
+    assert_equal "Push deliveries must be pinned to a resolved endpoint IP", error.message
+  end
+
+  private
+    def deliver_to(server, connection:)
+      WebPush.payload_send \
+        message: "",
+        endpoint: "https://#{PushServiceTestHelper::HOST}:#{server.port}/fcm/send/test123",
+        endpoint_ip: PushServiceTestHelper::IP,
+        p256dh: "", auth: "", vapid: {},
+        connection: connection,
+        urgency: "high"
+    end
 end
@@ -0,0 +1,53 @@
+require "test_helper"
+
+class WebPush::PoolTest < ActiveSupport::TestCase
+  setup do
+    stub_web_push_dns_resolution
+    @pool = Rails.configuration.x.web_push_pool
+    @payload = { title: "Designers", body: "Hello", path: Rails.application.routes.url_helpers.room_path(rooms(:designers)) }
+  end
+
+  test "every notification carries its subscriber's unread room count as the badge" do
+    memberships(:jason_pets).update! unread_at: Time.current
+    memberships(:kevin_hq).update! unread_at: Time.current
+    memberships(:kevin_david_and_kevin).update! unread_at: Time.current
+    subscriptions = Push::Subscription.where(user: users(:david, :jason, :kevin))
+    expected = subscriptions.to_h { |subscription| [ subscription.endpoint, subscription.user.memberships.unread.count ] }
+    assert_equal [ 0, 1, 2 ], expected.values.sort
+
+    badges = Concurrent::Hash.new
+    WebPush.stubs(:payload_send).with { |options| badges[options[:endpoint]] = JSON.parse(options[:message]).dig("options", "data", "badge") }
+    @pool.queue(@payload, subscriptions)
+    wait_for_deliveries(3)
+
+    assert_equal expected, badges
+  end
+
+  test "queueing counts unread rooms per batch, not per subscription" do
+    WebPush.stubs(:payload_send)
+
+    two = Push::Subscription.where(user: users(:david, :jason))
+    four = Push::Subscription.where(user: users(:david, :jason, :jz, :kevin))
+    assert_equal [ 2, 4 ], [ two.count, four.count ]
+
+    queries_for_two = count_queries { @pool.queue(@payload, two) }
+    queries_for_four = count_queries { @pool.queue(@payload, four) }
+    wait_for_deliveries(6)
+
+    assert_equal queries_for_two, queries_for_four
+  end
+
+  private
+    def wait_for_deliveries(count)
+      Timeout.timeout(2) { sleep 0.01 while @pool.delivery_pool.completed_task_count < count }
+    end
+
+    def count_queries(&block)
+      count = 0
+      counter = ->(*, payload) { count += 1 unless payload[:name] == "SCHEMA" || payload[:cached] }
+      ActiveRecord::Base.uncached do
+        ActiveSupport::Notifications.subscribed(counter, "sql.active_record", &block)
+      end
+      count
+    end
+end
@@ -19,12 +19,94 @@ class Message::AttachmentTest < ActiveSupport::TestCase
    assert_equal message.plain_text_body, "moon.jpg"
  end

+  test "creating a message keeps an image that can't be decoded" do
+    webp = Vips::Image.new_from_file(file_fixture("moon.jpg").to_s).webpsave_buffer
+    message = create_unreadable_attachment_message(webp.byteslice(0, webp.bytesize / 2), "broken.webp")
+
+    assert_equal "broken.webp", message.reload.attachment.filename.to_s
+    assert_nil message.attachment.representation(:thumb).image
+  end
+
+  test "creating a message keeps a video that can't be decoded" do
+    message = create_unreadable_attachment_message(file_fixture("alpha-centuri.mov").binread(64), "broken.mov")
+
+    assert_equal "broken.mov", message.reload.attachment.filename.to_s
+    assert_not message.attachment.preview(format: :webp).image.attached?
+  end
+
+  test "creating a message makes the video poster that the room shows" do
+    message = create_attachment_message("alpha-centuri.mov", "video/quicktime")
+
+    assert_no_difference -> { ActiveStorage::VariantRecord.count } do
+      message.reload.attachment.preview(:poster).processed
+    end
+  end
+
+  test "video previews are drawn by the previewer with a time limit, from a frame within the first 5 seconds" do
+    assert_includes ActiveStorage.previewers, TimeLimitedVideoPreviewer
+    assert_not_includes ActiveStorage.previewers, ActiveStorage::Previewer::VideoPreviewer
+    assert_includes ActiveStorage.video_preview_arguments, "gte(t\\,5)"
+  end
+
+  test "creating a message gives up on a video preview that takes longer than the time limit" do
+    started = nil
+    message = nil
+
+    with_ffmpeg_sleeping(5.seconds) do
+      stub_const(TimeLimitedVideoPreviewer, :TIME_LIMIT, 0.2) do
+        started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+        message = create_attachment_message("alpha-centuri.mov", "video/quicktime")
+      end
+    end
+
+    assert_operator Process.clock_gettime(Process::CLOCK_MONOTONIC) - started, :<, 2
+    assert_not message.reload.attachment.preview(format: :webp).image.attached?
+  end
+
+  test "creating a message makes no thumbnail of an image with more pixels than the limit" do
+    moon = Vips::Image.new_from_file(file_fixture("moon.jpg").to_s)
+
+    stub_const(Message::Attachment, :THUMBNAIL_MAX_PIXELS, moon.width * moon.height) do
+      assert create_attachment_message("moon.jpg", "image/jpeg").attachment.representation(:thumb).processed?
+    end
+
+    stub_const(Message::Attachment, :THUMBNAIL_MAX_PIXELS, moon.width * moon.height - 1) do
+      assert_not create_attachment_message("moon.jpg", "image/jpeg").attachment.representation(:thumb).processed?
+    end
+  end
+
+  test "creating a message makes no preview of a video with more pixels than the limit" do
+    stub_const(Message::Attachment, :THUMBNAIL_MAX_PIXELS, 1) do
+      message = create_attachment_message("alpha-centuri.mov", "video/quicktime")
+      assert_not message.reload.attachment.preview(format: :webp).image.attached?
+    end
+  end

  private
+    def with_ffmpeg_sleeping(duration)
+      Tempfile.create("ffmpeg") do |ffmpeg|
+        ffmpeg.write "#!/bin/sh\n[ \"$1\" = \"-version\" ] && exit 0\nexec sleep #{duration.to_i}\n"
+        ffmpeg.close
+        File.chmod 0o755, ffmpeg.path
+
+        previous, ActiveStorage.paths[:ffmpeg] = ActiveStorage.paths[:ffmpeg], ffmpeg.path
+        yield
+      ensure
+        ActiveStorage.paths[:ffmpeg] = previous
+      end
+    end
+
    def create_attachment_message(file, content_type)
      rooms(:hq).messages.create_with_attachment! \
        creator: users(:david),
        client_message_id: "message",
        attachment: fixture_file_upload(file, content_type)
    end
+
+    def create_unreadable_attachment_message(content, filename)
+      rooms(:hq).messages.create_with_attachment! \
+        creator: users(:david),
+        client_message_id: "message",
+        attachment: { io: StringIO.new(content), filename: filename }
+    end
 end
@@ -0,0 +1,16 @@
+require "test_helper"
+
+class Message::PaginationTest < ActiveSupport::TestCase
+  test "messages updated since a time are the newest page of them, in creation order" do
+    room = rooms(:watercooler)
+    first, second, third = room.messages.ordered.first(3)
+
+    travel 1.minute do
+      [ third, first, second ].each(&:touch)
+
+      stub_const(Message::Pagination, :PAGE_SIZE, 2) do
+        assert_equal [ second, third ], room.messages.page_updated_since(30.seconds.ago).to_a
+      end
+    end
+  end
+end
@@ -1,6 +1,9 @@
 require "test_helper"
+require "active_record/testing/query_assertions"

 class Message::SearchableTest < ActiveSupport::TestCase
+  include ActionDispatch::TestProcess, ActiveRecord::Assertions::QueryAssertions
+
  test "message body is indexed and searchable" do
    message = rooms(:designers).messages.create! body: "My hovercraft is full of eels", client_message_id: "earth", creator: users(:david)
    assert_equal [ message ], rooms(:designers).messages.search("eel")
@@ -12,6 +15,36 @@ class Message::SearchableTest < ActiveSupport::TestCase
    assert_equal [], rooms(:designers).messages.search("sharks")
  end

+  test "boosting a message leaves the search index alone, whether or not its body is loaded" do
+    attachment_message = rooms(:designers).messages.create! attachment: fixture_file_upload("moon.jpg", "image/jpeg"), client_message_id: "moon", creator: users(:david)
+
+    [ messages(:first), Message.with_rich_text_body.find(messages(:first).id), Message.with_rich_text_body.find(attachment_message.id) ].each do |message|
+      assert_no_queries_match(/message_search_index/) do
+        message.boosts.create!(content: "🦞", booster: users(:jason)).destroy!
+      end
+    end
+  end
+
+  test "saving a new body replaces the old words in the index" do
+    message = rooms(:designers).messages.create! body: "My hovercraft is full of eels", client_message_id: "earth", creator: users(:david)
+
+    Message.find(message.id).update! body: "My hovercraft is full of sharks"
+    assert_equal [ message ], rooms(:designers).messages.search("sharks")
+    assert_equal [], rooms(:designers).messages.search("eels")
+
+    Message.find(message.id).body.update! body: "My hovercraft is full of whales"
+    assert_equal [ message ], rooms(:designers).messages.search("whales")
+    assert_equal [], rooms(:designers).messages.search("sharks")
+  end
+
+  test "a new attachment replaces the old file name in the index" do
+    message = rooms(:designers).messages.create! attachment: fixture_file_upload("moon.jpg", "image/jpeg"), client_message_id: "moon", creator: users(:david)
+
+    Message.find(message.id).update! attachment: fixture_file_upload("pixel.bmp", "image/bmp")
+    assert_equal [ message ], rooms(:designers).messages.search("pixel")
+    assert_equal [], rooms(:designers).messages.search("moon")
+  end
+
  test "search results are returned in message order" do
    messages = [ "first cat", "second cat", "third cat", "cat cat cat" ].map do |body|
      rooms(:designers).messages.create! body: body, client_message_id: body, creator: users(:david)
@@ -20,6 +53,37 @@ class Message::SearchableTest < ActiveSupport::TestCase
    assert_equal messages, rooms(:designers).messages.search("cat")
  end

+  test "the last page of matches holds the newest ones, read off the index without sorting every match" do
+    messages = [ "first cat", "second cat", "third cat" ].map do |body|
+      rooms(:designers).messages.create! body: body, client_message_id: body, creator: users(:david)
+    end
+
+    queries = []
+    collect = ->(*, payload) { queries << payload[:sql] if payload[:sql].include?("message_search_index") }
+    page = ActiveSupport::Notifications.subscribed(collect, "sql.active_record") do
+      rooms(:designers).messages.search("cat").last_page_of_matches(2)
+    end
+
+    assert_equal messages.last(2), page
+    assert_no_match(/TEMP B-TREE/, Message.connection.select_rows("EXPLAIN QUERY PLAN #{queries.sole}").map(&:last).join(" | "))
+  end
+
+  test "bounded search keeps older reachable hits behind inaccessible matches" do
+    accessible = rooms(:designers).messages.create! body: "bounded sparse needle", creator: users(:david)
+    private_room = Room.create! name: "Private search history", type: "Rooms::Closed", creator: users(:jason)
+    now = Time.current
+    records = 1100.times.map do |i|
+      { room_id: private_room.id, creator_id: users(:jason).id, client_message_id: "private-#{i}", created_at: now, updated_at: now }
+    end
+    ids = Message.insert_all!(records, returning: %w[id]).rows.flatten
+    ids.each do |id|
+      Message.connection.execute Message.sanitize_sql([ "INSERT INTO message_search_index(rowid,body) VALUES (?,?)", id, "bounded sparse needle" ])
+    end
+    assert_equal [ accessible.id ], Message.search_reachable(users(:david), "bounded sparse").map(&:id)
+    rooms(:designers).memberships.where(user: users(:david)).delete_all
+    assert_empty Message.search_reachable(users(:david), "bounded sparse")
+  end
+
  test "rich text body is converted to plain text for indexing" do
    message = rooms(:designers).messages.create! body: "<span>My hovercraft is full of eels</span>", client_message_id: "earth", creator: users(:david)

@@ -65,6 +65,18 @@ class MessageTest < ActiveSupport::TestCase
    assert_equal room.messages.ordered.first(Message::Pagination::PAGE_SIZE).map(&:id), room.messages.first_page.map(&:id)
  end

+  test "paged? tells whether a room has more than a page of messages without counting them all" do
+    room = Rooms::Closed.create!(name: "Paging", creator: users(:david))
+    Message.insert_all Array.new(Message::Pagination::PAGE_SIZE) { |i| { room_id: room.id, creator_id: users(:david).id, client_message_id: "paging-#{i}" } }
+
+    assert_no_queries_match(/COUNT/i) do
+      assert_not room.messages.paged?
+
+      Message.insert_all [ { room_id: room.id, creator_id: users(:david).id, client_message_id: "paging-next" } ]
+      assert room.messages.paged?
+    end
+  end
+
  test "search reads quotes and operator words in the query as text" do
    message = rooms(:designers).messages.create!(body: "Say hi, NOT bye", client_message_id: "quoted", creator: users(:david))

@@ -125,6 +125,15 @@ class Push::SubscriptionTest < ActiveSupport::TestCase
    subscription.notification(title: "t", body: "b", path: "/").deliver
  end

+  test "the badge defaults to the subscriber's unread room count" do
+    memberships(:david_pets).update! unread_at: Time.current
+    memberships(:david_hq).update! unread_at: Time.current
+    subscription = build_subscription(endpoint: "https://fcm.googleapis.com/fcm/send/abc123")
+
+    WebPush.expects(:payload_send).with { |options| JSON.parse(options[:message]).dig("options", "data", "badge") == 2 }
+    subscription.notification(title: "t", body: "b", path: "/").deliver
+  end
+
  test "delivery sends with the pinned endpoint_ip" do
    subscription = build_subscription(endpoint: "https://fcm.googleapis.com/fcm/send/abc123")

@@ -63,7 +63,44 @@ class Room::PushTest < ActiveSupport::TestCase
    end
  end

+  test "does not notify banned users" do
+    users(:kevin).ban
+
+    assert_not_includes pushed_users { post_to_designers_mentioning_kevin }, users(:kevin)
+  end
+
+  test "notifies users again once they're unbanned" do
+    users(:kevin).ban
+    users(:kevin).unban
+
+    assert_includes pushed_users { post_to_designers_mentioning_kevin }, users(:kevin)
+  end
+
+  test "does not notify banned users of direct messages" do
+    users(:kevin).ban
+
+    pushed = pushed_users do
+      rooms(:david_and_kevin).messages.create! body: "Just between us", client_message_id: "earth", creator: users(:david)
+    end
+    assert_not_includes pushed, users(:kevin)
+  end
+
+  test "notifies active users mentioned" do
+    assert_includes pushed_users { post_to_designers_mentioning_kevin }, users(:kevin)
+  end
+
  private
+    def post_to_designers_mentioning_kevin
+      rooms(:designers).messages.create! body: "Hey #{mention_attachment_for(:kevin)}", client_message_id: "earth", creator: users(:david)
+    end
+
+    def pushed_users(&block)
+      queued = []
+      Rails.configuration.x.web_push_pool.stubs(:queue).with { |_payload, subscriptions| queued.concat subscriptions.to_a }
+      perform_enqueued_jobs(only: Room::PushMessageJob, &block)
+      queued.map(&:user).uniq
+    end
+
    def wait_for_web_push_delivery_pool_tasks(count)
      wait_for_pool_tasks(Rails.configuration.x.web_push_pool.delivery_pool, count)
    end
@@ -1,6 +1,8 @@
 require "test_helper"

 class RoomTest < ActiveSupport::TestCase
+  include ActionCable::TestHelper
+
  test "grant membership to user" do
    rooms(:watercooler).memberships.grant_to(users(:kevin))
    assert rooms(:watercooler).users.include?(users(:kevin))
@@ -30,8 +32,67 @@ class RoomTest < ActiveSupport::TestCase
    assert Rooms::Closed.new.closed?
  end

+  test "an open room destroyed later lets no one in who joins the account before the job runs" do
+    room = rooms(:pets)
+
+    room.destroy_later
+    newcomer = User.create!(name: "Newcomer", email_address: "newcomer@example.com", password: "secret123456")
+
+    assert_not newcomer.memberships.exists?(room_id: room.id)
+    perform_enqueued_jobs only: Room::DestroyJob
+    assert_not Room.exists?(room.id)
+  end
+
+  test "a room destroyed later resets the connections of its former members, as revoking their memberships would" do
+    room = rooms(:designers)
+    former_members = room.users.to_a
+    assert former_members.many?
+
+    room.destroy_later
+    perform_enqueued_jobs only: Room::DestroyJob
+
+    former_members.each do |member|
+      assert_broadcast_on "action_cable/#{member.to_gid_param}", { type: "disconnect", reconnect: true }
+    end
+  end
+
+  test "destroying one message at a time leaves nothing of the room behind" do
+    room = rooms(:designers)
+    searchable = room.messages.create!(body: "Kept in the search index", creator: users(:david))
+    searchable.attachment.attach io: StringIO.new("hello"), filename: "hello.txt", content_type: "text/plain"
+    message_ids = room.messages.ids
+    assert Boost.where(message_id: message_ids).exists?
+    assert_equal 1, search_index_rows(searchable)
+
+    room.destroy_one_message_at_a_time
+
+    assert_not Room.exists?(room.id)
+    assert_empty Message.where(id: message_ids)
+    assert_empty Boost.where(message_id: message_ids)
+    assert_empty ActionText::RichText.where(record_type: "Message", record_id: message_ids)
+    assert_empty ActiveStorage::Attachment.where(record_type: "Message", record_id: message_ids)
+    assert_enqueued_jobs 1, only: ActiveStorage::PurgeJob
+    assert_equal 0, search_index_rows(searchable)
+  end
+
+  test "each message is destroyed in its own transaction, so other writes get through in between" do
+    room = rooms(:designers)
+    messages = room.messages.count
+    transactions = 0
+    count_transactions = ->(*, payload) { transactions += 1 if payload[:sql].start_with?("RELEASE SAVEPOINT") }
+
+    ActiveSupport::Notifications.subscribed(count_transactions, "sql.active_record") { room.destroy_one_message_at_a_time }
+
+    assert_equal messages + 1, transactions
+  end
+
  test "default involvement for new users" do
    room = Rooms::Closed.create_for({ name: "Hello!", creator: users(:david) }, users: [ users(:kevin), users(:david) ])
    assert room.memberships.all? { |m| m.involved_in_mentions? }
  end
+
+  private
+    def search_index_rows(message)
+      Message.connection.select_value("select count(*) from message_search_index where rowid = #{message.id}")
+    end
 end
@@ -18,4 +18,33 @@ class Rooms::DirectTest < ActiveSupport::TestCase
    room = Rooms::Direct.find_or_create_for([ users(:david), users(:kevin) ])
    assert room.memberships.all? { |m| m.involved_in_everything? }
  end
+
+  test "find the room with exactly the same users" do
+    Current.user = users(:david)
+    group = Rooms::Direct.find_or_create_for([ users(:david), users(:kevin), users(:jason) ])
+
+    assert_equal rooms(:david_and_kevin), Rooms::Direct.find_or_create_for([ users(:kevin), users(:david) ])
+    assert_equal group, Rooms::Direct.find_or_create_for(User.where(id: [ users(:jason), users(:kevin), users(:david) ]))
+    assert_equal [ users(:david), users(:jz) ].to_set, Rooms::Direct.find_or_create_for([ users(:david), users(:jz) ]).users.to_set
+    assert_equal [ users(:jason), users(:kevin) ].to_set, Rooms::Direct.find_or_create_for([ users(:jason), users(:kevin) ]).users.to_set
+  end
+
+  test "finding a room takes the same queries however many direct rooms the account has" do
+    Current.user = users(:david)
+    Rooms::Direct.create_for({}, users: [ users(:david), users(:jz) ])
+    before = queries_during { Rooms::Direct.find_or_create_for(User.where(id: [ users(:david), users(:jz) ])) }
+
+    5.times { |i| Rooms::Direct.create_for({}, users: [ users(:jason), User.create!(name: "Guest #{i}", email_address: "guest#{i}@example.com", password: "secret123456") ]) }
+    Rooms::Direct.create_for({}, users: [ users(:kevin), users(:jz) ])
+
+    assert_equal before, queries_during { Rooms::Direct.find_or_create_for(User.where(id: [ users(:kevin), users(:jz) ])) }
+  end
+
+  private
+    def queries_during(&block)
+      count = 0
+      counter = ->(*, payload) { count += 1 unless payload[:name] == "SCHEMA" || payload[:cached] }
+      ActiveSupport::Notifications.subscribed(counter, "sql.active_record", &block)
+      count
+    end
 end
@@ -0,0 +1,10 @@
+require "test_helper"
+
+class User::BannableTest < ActiveSupport::TestCase
+  test "banned users are visible to administrators only" do
+    users(:kevin).banned!
+
+    assert_includes User.visible_to(users(:david)), users(:kevin)
+    assert_not_includes User.visible_to(users(:jz)), users(:kevin)
+  end
+end
@@ -12,6 +12,18 @@ class UserTest < ActiveSupport::TestCase
    end
  end

+  test "memberships granted to the open rooms are like any other" do
+    freeze_time
+    user = create_new_user
+
+    assert_equal Rooms::Open.ids.sort, user.memberships.pluck(:room_id).sort
+    user.memberships.each do |membership|
+      assert membership.involved_in_mentions?
+      assert_equal Time.current, membership.created_at
+      assert_equal Time.current, membership.updated_at
+    end
+  end
+
  test "deactivating a user deletes push subscriptions, searches, memberships for non-direct rooms, and changes their email address" do
    assert_difference -> { Membership.count }, -users(:david).memberships.without_direct_rooms.count do
    assert_difference -> { Push::Subscription.count }, -users(:david).push_subscriptions.count do
@@ -41,7 +41,7 @@ class WebhookTest < ActiveSupport::TestCase

  test "delivery with error reply" do
    assert_no_difference -> { Message.count } do
-      WebMock.stub_request(:post, webhooks(:bender).url).to_return(status: 500, body: "Internal Error!", headers: {})
+      WebMock.stub_request(:post, webhooks(:bender).url).to_return(status: 500, body: "<h1>Internal Error!</h1>", headers: { "Content-Type" => "text/html" })
      response = webhooks(:bender).deliver(messages(:first))
    end
  end
@@ -0,0 +1,150 @@
+require "application_system_test_case"
+
+# Visual proof for GHSA-3v99-4vxh-xg84 under the server-id fix.
+#
+# Before the fix a message rendered with the DOM id "message_<client_message_id>",
+# where the client_message_id is chosen by the browser. create/update/destroy Turbo
+# Stream broadcasts targeted that DOM id and a browser resolves it to the FIRST
+# matching element, so an attacker (a non-admin room member) who reused the victim's
+# client_message_id could displace the victim's message in every connected browser.
+#
+# This fix derives every message DOM id from the record's primary key instead, so two
+# distinct records can never share a DOM id no matter what client_message_id they carry.
+# The attacker's colliding create still succeeds (there is no uniqueness constraint and
+# none is needed) but lands as its own element with its own PK-based id; editing it
+# broadcasts to the attacker's own presentation id and cannot touch the victim.
+#
+# The test drives two real headless-Chromium sessions (victim + attacker) against one
+# server and screenshots the outcome to tmp/screenshots/.
+class MessageIdSpoofingTest < ApplicationSystemTestCase
+  SCREENSHOT_DIR = Rails.root.join("tmp/screenshots")
+
+  setup do
+    FileUtils.mkdir_p(SCREENSHOT_DIR)
+  end
+
+  test "a reused client_message_id cannot hijack the victim message" do
+    room = rooms(:designers)
+
+    # Victim (jz) posts a real message through the composer and keeps the room open.
+    sign_in "jz@37signals.com"
+    join_room room
+    send_message "Victim's real message"
+    assert_selector ".message[data-message-id]", text: "Victim's real message"
+
+    # The browser chose the client_message_id; read it (and the record id) back
+    # from the persisted message the victim just posted.
+    victim_message = room.messages.where(creator: users(:jz)).order(:created_at).last
+    victim_client_id = victim_message.client_message_id
+    victim_db_id = victim_message.id
+
+    save_screenshot(SCREENSHOT_DIR.join("spoof-before.png"))
+
+    # Attacker (kevin), a different non-admin member of the same room, posts a message
+    # reusing the victim's client_message_id, then edits their own message.
+    attacker_status = nil
+    using_session("attacker") do
+      sign_in "kevin@37signals.com"
+      join_room room
+      attacker_status = post_message_via_browser(room, "HIJACKED BY ATTACKER", victim_client_id)
+      assert (200..299).include?(attacker_status), "the colliding create should still succeed (got #{attacker_status})"
+
+      attacker_db_id = find(".message[data-message-id]", text: "HIJACKED BY ATTACKER")["data-message-id"]
+      edit_message_via_browser(room, attacker_db_id, "EDITED BY ATTACKER")
+    end
+
+    attacker_message = room.messages.where(creator: users(:kevin)).order(:created_at).last
+    attacker_db_id = attacker_message.id
+
+    # Both messages carry the SAME client_message_id...
+    assert_equal victim_client_id, attacker_message.client_message_id
+    assert_not_equal victim_db_id, attacker_db_id
+
+    # ...but in the victim's browser they render as two DISTINCT elements. Elements
+    # are located by record id (data-message-id), not by DOM id, so this fails on
+    # the vulnerable code for the real reason: the victim's element is gone. Wait
+    # for the attacker's edit to arrive in the victim's browser, then confirm the
+    # victim's own message is untouched and still under the victim's name.
+    assert_selector message_selector(attacker_db_id), text: "EDITED BY ATTACKER"
+    assert_selector "#{message_selector(victim_db_id)} .message__body", text: "Victim's real message"
+    assert_selector "#{message_selector(victim_db_id)} .message__author", text: "JZ"
+    assert_selector ".message[data-message-id]", count: room.messages.count
+
+    # The victim's element never took on the attacker's text.
+    assert_no_selector "#{message_selector(victim_db_id)} .message__body", text: /ATTACKER/
+
+    # And the victim's record is untouched on the server.
+    assert_equal "Victim's real message", victim_message.reload.body.to_plain_text.strip
+
+    save_screenshot(SCREENSHOT_DIR.join("spoof-after-fixed.png"))
+  end
+
+  test "a distinct client_message_id never alters the victim message" do
+    room = rooms(:designers)
+
+    sign_in "jz@37signals.com"
+    join_room room
+    send_message "Victim's untouched message"
+    assert_selector ".message[data-message-id]", text: "Victim's untouched message"
+
+    victim_db_id = room.messages.where(creator: users(:jz)).order(:created_at).last.id
+    attacker_db_id = nil
+
+    using_session("attacker") do
+      sign_in "kevin@37signals.com"
+      join_room room
+      status = post_message_via_browser(room, "Attacker's own message", "a-distinct-id")
+      assert (200..299).include?(status), "a non-colliding create should succeed (got #{status})"
+
+      attacker_db_id = find(".message[data-message-id]", text: "Attacker's own message")["data-message-id"]
+      edit_message_via_browser(room, attacker_db_id, "EDITED BY ATTACKER")
+    end
+
+    # The attacker's edit lands only on the attacker's own message. The victim's
+    # message is untouched.
+    assert_selector message_selector(attacker_db_id), text: "EDITED BY ATTACKER"
+    assert_selector "#{message_selector(victim_db_id)} .message__body", text: "Victim's untouched message"
+    assert_no_selector "#{message_selector(victim_db_id)} .message__body", text: "EDITED BY ATTACKER"
+  end
+
+  private
+    def message_selector(message_id)
+      ".message[data-message-id='#{message_id}']"
+    end
+
+    # POST a message through the attacker's live authenticated browser session,
+    # forcing a chosen client_message_id. Synchronous XHR so we get the real HTTP
+    # status back. Returns the status code.
+    def post_message_via_browser(room, body, client_message_id)
+      page.evaluate_script(<<~JS)
+        (function() {
+          var xhr = new XMLHttpRequest();
+          xhr.open('POST', '/rooms/#{room.id}/messages', false);
+          var token = document.querySelector('meta[name="csrf-token"]');
+          if (token) { xhr.setRequestHeader('X-CSRF-Token', token.content); }
+          xhr.setRequestHeader('Accept', 'text/vnd.turbo-stream.html');
+          var fd = new FormData();
+          fd.append('message[body]', #{body.to_json});
+          fd.append('message[client_message_id]', #{client_message_id.to_json});
+          xhr.send(fd);
+          return xhr.status;
+        })()
+      JS
+    end
+
+    # PATCH the attacker's own message through their live browser session. Fired
+    # async (the caller then waits on the resulting DOM broadcast); the update's
+    # replace broadcast targets the attacker's own presentation DOM id.
+    def edit_message_via_browser(room, message_id, body)
+      page.execute_script(<<~JS)
+        (function() {
+          var fd = new FormData();
+          fd.append('message[body]', #{body.to_json});
+          var headers = { 'Accept': 'text/html' };
+          var token = document.querySelector('meta[name="csrf-token"]');
+          if (token) { headers['X-CSRF-Token'] = token.content; }
+          fetch('/rooms/#{room.id}/messages/#{message_id}', { method: 'PATCH', headers: headers, body: fd });
+        })()
+      JS
+    end
+end
@@ -15,6 +15,11 @@ class SendingMessagesTest < ApplicationSystemTestCase
    join_room rooms(:designers)
    send_message "Is this thing on?"

+    # The optimistic pending message is reconciled with the server broadcast even
+    # though they no longer share a DOM id: exactly one element, no duplicate.
+    assert_selector ".message__body", text: "Is this thing on?", count: 1
+    assert_no_selector "[data-pending-message]"
+
    using_session("Kevin") do
      join_room rooms(:designers)
      assert_message_text "Is this thing on?"
@@ -15,8 +15,11 @@ class UnreadRoomsTest < ApplicationSystemTestCase
    using_session("Kevin") do
      sign_in "kevin@37signals.com"
      join_room designers_room
-      send_message("Hello!!")
-      send_message("Talking to myself?")
+
+      perform_enqueued_jobs only: Message::BroadcastUnreadRoomJob do
+        send_message("Hello!!")
+        send_message("Talking to myself?")
+      end
    end

    assert_room_unread designers_room
@@ -24,4 +27,51 @@ class UnreadRoomsTest < ApplicationSystemTestCase
    join_room designers_room
    assert_room_read designers_room
  end
+
+  test "a notice about a message from before the room was read in another tab doesn't mark it unread again" do
+    using_session("Kevin in HQ") do
+      sign_in "kevin@37signals.com"
+      join_room rooms(:hq)
+      broadcast_unread_notice rooms(:designers), to: users(:kevin), at: Time.current
+      assert_room_unread rooms(:designers)
+    end
+
+    sent_before_reading = Time.current
+    using_session("Kevin in Designers") do
+      sign_in "kevin@37signals.com"
+      join_room rooms(:designers)
+    end
+
+    using_session("Kevin in HQ") do
+      assert_room_read rooms(:designers)
+
+      broadcast_unread_notice rooms(:designers), to: users(:kevin), at: sent_before_reading
+      broadcast_unread_notice rooms(:bender_and_kevin), to: users(:kevin), at: Time.current
+      assert_selector "#" + dom_id(rooms(:bender_and_kevin), :list) + ".unread"
+      assert_room_read rooms(:designers)
+
+      broadcast_unread_notice rooms(:designers), to: users(:kevin), at: Time.current
+      assert_room_unread rooms(:designers)
+    end
+  end
+
+  test "a message after a read in the same second marks the room unread" do
+    room = rooms(:designers)
+    user = users(:jz)
+    join_room rooms(:hq)
+    read_at = Time.current.change(usec: 100_000)
+
+    broadcast_unread_notice room, to: user, at: read_at - 1
+    assert_room_unread room
+    ActionCable.server.broadcast "user_#{user.id}_reads", { room_id: room.id, at: read_at.to_f }
+    assert_room_read room
+
+    broadcast_unread_notice room, to: user, at: read_at + 0.2
+    assert_room_unread room
+  end
+
+  private
+    def broadcast_unread_notice(room, to:, at:)
+      ActionCable.server.broadcast UnreadRoomsChannel.stream_name_for(to.id), { roomId: room.id, at: at.to_f }
+    end
 end
@@ -2,7 +2,6 @@ ENV["RAILS_ENV"] ||= "test"
 require_relative "../config/environment"

 require "rails/test_help"
-require "minitest/unit"
 require "mocha/minitest"
 require "webmock/minitest"
 require "turbo/broadcastable/test_helper"
@@ -0,0 +1,121 @@
+module PushServiceTestHelper
+  # A push service on 127.0.0.1 for delivery tests: TLS with a certificate for HOST from a test CA that the
+  # process trusts, HTTP/1.1 keep-alive. HOST doesn't resolve, so a delivery that looked it up instead of
+  # using its pinned address would fail.
+  HOST = "push.test"
+  IP = "127.0.0.1"
+
+  class Server
+    attr_reader :port
+
+    # status: the answer to every request.
+    # hang_up_after_response: :fin or :close_notify to hang up after each answer, as an idle timeout would.
+    # drop_request: ->(number) whether to hang up after reading that request, without answering it.
+    # certificate: ->(connection) the name its certificate is for, HOST unless told otherwise.
+    def initialize(status: "201 Created", hang_up_after_response: nil, drop_request: ->(_) { false }, certificate: ->(_) { HOST })
+      @status, @hang_up_after_response, @drop_request, @certificate = status, hang_up_after_response, drop_request, certificate
+      @connections, @requests, @closed = 0, [], Queue.new
+      @mutex = Mutex.new
+      @listener = TCPServer.new(IP, 0)
+      @port = @listener.addr[1]
+      @thread = Thread.new { loop { serve(@listener.accept) } rescue IOError }
+    end
+
+    def connections = @mutex.synchronize { @connections }
+    def requests = @mutex.synchronize { @requests.dup }
+
+    # Whether a connection to it closed, waiting for that up to the given seconds.
+    def hung_up?(within: 5)
+      !!@closed.pop(timeout: within)
+    end
+
+    def stop
+      @listener.close
+      @thread.join
+    end
+
+    private
+      def serve(socket)
+        connection = @mutex.synchronize { @connections += 1 }
+        Thread.new do
+          tls = OpenSSL::SSL::SSLSocket.new(socket, PushServiceTestHelper.context_for(@certificate.(connection))).tap(&:accept)
+          while (request_line = tls.gets("\r\n"))
+            length = 0
+            while (header = tls.gets("\r\n")) != "\r\n"
+              length = header.split(":", 2).last.to_i if header.downcase.start_with?("content-length:")
+            end
+            tls.read(length)
+            number = @mutex.synchronize { @requests << request_line.split[1]; @requests.size }
+            break if @drop_request.(number)
+            tls.write "HTTP/1.1 #{@status}\r\nContent-Length: 0\r\n\r\n"
+            if @hang_up_after_response
+              tls.close if @hang_up_after_response == :close_notify
+              break
+            end
+          end
+        rescue IOError, SystemCallError, OpenSSL::SSL::SSLError
+        ensure
+          socket.close
+          @closed << true
+        end
+      end
+  end
+
+  class << self
+    def context_for(name)
+      @contexts ||= {}
+      @contexts[name] ||= OpenSSL::SSL::SSLContext.new.tap do |context|
+        context.key = OpenSSL::PKey::EC.generate("prime256v1")
+        context.cert = sign(subject: "CN=#{name}", key: context.key, ca: ca, ca_key: ca_key) do |extensions|
+          [ extensions.create_extension("subjectAltName", "DNS:#{name}") ]
+        end
+      end
+    end
+
+    private
+      def ca_key
+        @ca_key ||= OpenSSL::PKey::EC.generate("prime256v1")
+      end
+
+      def ca
+        @ca ||= sign(subject: "CN=Push service test CA", key: ca_key, ca_key: ca_key) do |extensions|
+          [ extensions.create_extension("basicConstraints", "CA:TRUE", true), extensions.create_extension("keyUsage", "keyCertSign", true) ]
+        end.tap { OpenSSL::SSL::SSLContext::DEFAULT_CERT_STORE.add_cert(it) }
+      end
+
+      def sign(subject:, key:, ca_key:, ca: nil)
+        OpenSSL::X509::Certificate.new.tap do |certificate|
+          certificate.version, certificate.serial = 2, SecureRandom.random_number(1 << 64)
+          certificate.subject = OpenSSL::X509::Name.parse(subject)
+          certificate.issuer = ca ? ca.subject : certificate.subject
+          certificate.public_key = key
+          certificate.not_before, certificate.not_after = 1.minute.ago, 1.hour.from_now
+          extensions = OpenSSL::X509::ExtensionFactory.new(ca || certificate, certificate)
+          yield(extensions).each { certificate.add_extension(it) }
+          certificate.sign(ca_key, "SHA256")
+        end
+      end
+  end
+
+  private
+    def with_push_service(**options)
+      WebMock.disable!
+      server = Server.new(**options)
+      yield server
+    ensure
+      server&.stop
+      WebMock.enable!
+    end
+
+    # Built as WebPush::PersistentRequest builds it for a delivery.
+    def pinned_connection(server, ip = IP)
+      Net::HTTP.new(HOST, server.port, nil).tap do |http|
+        http.ipaddr = ip
+        http.use_ssl = true
+      end
+    end
+
+    def push_request(path = "/push")
+      Net::HTTP::Post.new(path).tap { it.body = "payload" }
+    end
+end