Implemented query caching for searches

This commit implements a possibly multi-user cache for long running
queries (currently the threshold is 0.4 seconds)
We use the max (message id, topic id) as a marker in the cache keys,
because these lookups are cheap. This means the cache remains valid for
multiple users as long as no new message arrives matching the search
criteria, and at time we automatically generate a new cache entry, while
existing views can still use the old cache record.

Author: dutow

app/controllers/topics_controller.rb

@@ -90,16 +90,19 @@ def read_all
   def search
     @search_query = params[:q].to_s.strip
 
-    base_query = topics_base_query(search_query: @search_query)
-
-    apply_cursor_pagination(base_query)
-    @new_topics_count = 0
+    if @search_query.present?
+      load_cached_search_results
+    else
+      base_query = topics_base_query(search_query: @search_query)
+      apply_cursor_pagination(base_query)
+      @new_topics_count = 0
+    end
 
     preload_participation_flags if user_signed_in?
 
     respond_to do |format|
       format.html
-      format.turbo_stream { render :index }
+      format.turbo_stream { render :search }
     end
   end
 
@@ -593,4 +596,79 @@ def load_notes
       @notes_by_message[key] << note
     end
   end
+
+  def load_cached_search_results
+    @viewing_since = viewing_since_param
+    longpage = params[:longpage].to_i
+    cache = SearchResultCache.new(query: @search_query, scope: "title_body", viewing_since: @viewing_since, longpage: longpage)
+
+    result = cache.fetch do |limit, offset|
+      build_search_query(@search_query)
+        .joins(:messages)
+        .where(messages: { created_at: ..@viewing_since })
+        .group('topics.id')
+        .select('topics.id, topics.creator_id, MAX(messages.created_at) as last_activity')
+        .order('MAX(messages.created_at) DESC, topics.id DESC')
+        .limit(limit)
+        .offset(offset)
+        .load
+    end
+
+    entries = result[:entries] || []
+    sliced = slice_cached_entries(entries, params[:cursor])
+
+    if sliced[:entries].empty? && entries.size >= SearchResultCache::LONGPAGE_SIZE
+      longpage += 1
+      cache = SearchResultCache.new(query: @search_query, scope: "title_body", viewing_since: @viewing_since, longpage: longpage)
+      next_result = cache.fetch do |limit, offset|
+        build_search_query(@search_query)
+          .joins(:messages)
+          .where(messages: { created_at: ..@viewing_since })
+          .group('topics.id')
+          .select('topics.id, topics.creator_id, MAX(messages.created_at) as last_activity')
+          .order('MAX(messages.created_at) DESC, topics.id DESC')
+          .limit(limit)
+          .offset(offset)
+          .load
+      end
+      entries = next_result[:entries] || []
+      sliced = slice_cached_entries(entries, params[:cursor])
+    end
+
+    @current_longpage = longpage
+    @topics = hydrate_topics_from_entries(sliced[:entries])
+    @topics = [] unless @topics
+    @new_topics_count = 0
+  end
+
+  def slice_cached_entries(entries, cursor_param)
+    return { entries: entries.first(25) } unless cursor_param.present?
+
+    cursor_time_str, cursor_id_str = cursor_param.split('_')
+    cursor_time = Time.zone.parse(cursor_time_str)
+    cursor_id = cursor_id_str.to_i
+
+    start_index = entries.find_index do |entry|
+      entry_time = entry[:last_activity]
+      next false unless entry_time
+      (entry_time < cursor_time) || (entry_time == cursor_time && entry[:id].to_i < cursor_id)
+    end
+
+    start_index ||= entries.size
+    { entries: entries.drop(start_index).first(25) }
+  end
+
+  def hydrate_topics_from_entries(entries)
+    ids = entries.map { |e| e[:id] }
+    return [] if ids.empty?
+
+    topics_map = Topic.includes(:creator).where(id: ids).index_by(&:id)
+    entries.filter_map do |entry|
+      topic = topics_map[entry[:id]]
+      next unless topic
+      last_activity = entry[:last_activity]
+      topic.define_singleton_method(:last_activity) { last_activity }
+      topic
+    end
+  end
 end

app/services/search_result_cache.rb

@@ -0,0 +1,92 @@
+require "digest"
+
+class SearchResultCache
+  CACHE_VERSION = "v1"
+  LONGPAGE_SIZE = 1000
+  CACHE_TTL = 6.hours
+  CACHE_THRESHOLD_SECONDS = 0.4 
+
+  def initialize(query:, scope:, viewing_since:, longpage: 0, cache: Rails.cache)
+    @query = query.to_s
+    @scope = scope.to_s
+    @viewing_since = viewing_since
+    @longpage = [longpage.to_i, 0].max
+    @cache = cache
+  end
+
+  def fetch
+    watermarks = compute_watermarks
+    cached = read_cached(watermarks)
+    return cached if cached
+
+    started_at = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+    entries = yield(LONGPAGE_SIZE, offset_for_longpage)
+    duration = Process.clock_gettime(Process::CLOCK_MONOTONIC) - started_at
+
+    payload = serialize_entries(entries).merge(watermarks:, cached_at: Time.current)
+
+    if duration > CACHE_THRESHOLD_SECONDS
+      write_cached(watermarks, payload)
+    end
+
+    payload
+  end
+
+  private
+
+  def offset_for_longpage
+    LONGPAGE_SIZE * @longpage
+  end
+
+  def read_cached(watermarks)
+    @cache.read(cache_key(watermarks))
+  end
+
+  def write_cached(watermarks, payload)
+    @cache.write(cache_key(watermarks), payload, expires_in: CACHE_TTL)
+  end
+
+  def cache_key(watermarks)
+    watermark_part = watermarks.sort_by { |k, _| k.to_s }.map { |k, v| "#{k}=#{v || 0}" }.join(":")
+    query_hash = Digest::SHA1.hexdigest(@query)
+    key  = [
+      "search",
+      CACHE_VERSION,
+      "scope", @scope,
+      "q", query_hash,
+      "wm", watermark_part,
+      "lp", @longpage
+    ].join(":")
+    File.open("cachelog.txt", 'a') do |f|
+      f.puts "CACHE KEY: #{key}"
+    end
+    key
+  end
+
+  def serialize_entries(entries)
+    {
+      entries: entries.map do |row|
+        {
+          id: row.id,
+          last_activity: row.try(:last_activity)&.to_time || row.try(:created_at)&.to_time
+        }
+      end
+    }
+  end
+
+  def compute_watermarks
+    case @scope
+    when "title_body"
+      pattern = "%#{ActiveRecord::Base.sanitize_sql_like(@query)}%"
+      title_max = Topic.where("title ILIKE ?", pattern)
+                       .where("created_at <= ?", @viewing_since)
+                       .maximum(:id) || 0
+      body_max = Message.where("body ILIKE ?", pattern)
+                        .where("created_at <= ?", @viewing_since)
+                        .maximum(:id) || 0
+      { title_max_id: title_max, body_max_id: body_max }
+    else
+      {}
+    end
+  end
+end

app/views/topics/search.html.slim

@@ -21,7 +21,7 @@
       - if @topics.size == 25
         - last_topic = @topics.last
         - cursor = "#{last_topic.last_activity.iso8601}_#{last_topic.id}"
-        = turbo_frame_tag "pagination", src: search_topics_path(q: @search_query, cursor: cursor, viewing_since: @viewing_since.iso8601, format: :turbo_stream), loading: :lazy do
+        = turbo_frame_tag "pagination", src: search_topics_path(q: @search_query, cursor: cursor, viewing_since: @viewing_since.iso8601, longpage: @current_longpage, format: :turbo_stream), loading: :lazy do
           .loading-indicator Loading more results...
   - else
     .no-results

app/views/topics/search.turbo_stream.slim

@@ -0,0 +1,9 @@
+= turbo_stream.append "topics" do
+  = render partial: "topics", locals: { topics: @topics }
+
+- if @topics.size == 25
+  - last_topic = @topics.last
+  - cursor = "#{last_topic.last_activity.iso8601}_#{last_topic.id}"
+  = turbo_stream.replace "pagination" do
+    = turbo_frame_tag "pagination", src: search_topics_path(q: @search_query, cursor: cursor, viewing_since: @viewing_since.iso8601, longpage: @current_longpage, format: :turbo_stream), loading: :lazy do
+      .loading-indicator Loading more results...