Swimlane view: compare streams side by side in the browser

ruby_event_store-browser/lib/ruby_event_store/browser.rb

@@ -22,6 +22,7 @@ def self.fingerprint(name)
 end
 
 require_relative "browser/get_events_from_stream"
+require_relative "browser/get_events_from_streams"
 require_relative "browser/urls"
 require_relative "browser/router"
 require_relative "browser/renderer"

ruby_event_store-browser/lib/ruby_event_store/browser/app.rb

@@ -162,6 +162,41 @@ def call(env)
                )
         end
 
+        router.add_route("GET", "/swimlane") do |params, urls|
+          stream_names, sort = swimlane_params(params)
+          reader = GetEventsFromStreams.new(event_store: event_store, stream_names: stream_names, sort: sort)
+          html render(
+                 "swimlane/show",
+                 urls: urls,
+                 stream_names: stream_names,
+                 events: reader.events,
+                 sort: sort,
+                 more_url: (urls.swimlane_more_url(stream_names, reader.next_cursor, sort) if reader.more?),
+               )
+        end
+
+        router.add_route("GET", "/swimlane/more") do |params, urls|
+          stream_names, sort = swimlane_params(params)
+          reader =
+            GetEventsFromStreams.new(
+              event_store: event_store,
+              stream_names: stream_names,
+              cursor: params["cursor"],
+              sort: sort,
+            )
+          json(
+            html:
+              Renderer.new.render(
+                "swimlane/_rows",
+                urls: urls,
+                stream_names: stream_names,
+                events: reader.events,
+                sort: sort,
+              ),
+            more_url: (urls.swimlane_more_url(stream_names, reader.next_cursor, sort) if reader.more?),
+          )
+        end
+
         extensions.each do |extension|
           extension.register_routes(router, ExtensionContext.new(event_store, method(:extension_stylesheets), method(:extension_scripts)))
         end
@@ -223,6 +258,16 @@ def html(body)
         [200, { "content-type" => "text/html;charset=utf-8" }, [body]]
       end
 
+      def json(body)
+        [200, { "content-type" => "application/json" }, [JSON.generate(body)]]
+      end
+
+      def swimlane_params(params)
+        stream_names = Array(params["streams"]).reject { |name| name.nil? || name.empty? }.uniq
+        sort = ("as_of" if params["sort"] == "as_of")
+        [stream_names, sort]
+      end
+
       def not_found(urls)
         renderer = Renderer.new
         content = renderer.render("not_found")

ruby_event_store-browser/lib/ruby_event_store/browser/get_events_from_streams.rb

@@ -0,0 +1,89 @@
+# frozen_string_literal: true
+
+module RubyEventStore
+  module Browser
+    GetEventsFromStreams =
+      Struct.new(:event_store, :stream_names, :cursor, :sort, :count, keyword_init: true) do
+        def initialize(event_store:, stream_names:, cursor: nil, sort: nil, count: PAGE_SIZE)
+          super(event_store: event_store, stream_names: stream_names, cursor: cursor, sort: sort, count: count)
+        end
+
+        def events
+          @events ||= build_page
+        end
+
+        def more?
+          events
+          @more
+        end
+
+        def next_cursor
+          events.last && time_of(events.last.last).iso8601(TIMESTAMP_PRECISION)
+        end
+
+        private
+
+        def build_page
+          rows = stream_names.flat_map { |name| chunks[name].map { |event| [name, event] } }
+          @more = full_chunks.any?
+          return group(rows) unless @more
+
+          boundary = full_chunks.map { |name| time_of(chunks[name].last) }.max
+          page_rows = rows.select { |_, event| time_of(event) >= boundary }
+          group(page_rows + drained_rows(boundary))
+        end
+
+        def chunks
+          @chunks ||= stream_names.to_h { |name| [name, read_chunk(name)] }
+        end
+
+        def read_chunk(name)
+          scope = time_sorted(stream_scope(name)).backward.limit(count)
+          scope = scope.older_than(cursor_time) if cursor_time
+          scope.to_a
+        end
+
+        def full_chunks
+          stream_names.select { |name| chunks[name].size == count }
+        end
+
+        def drained_rows(boundary)
+          full_chunks
+            .select { |name| time_of(chunks[name].last) == boundary }
+            .flat_map { |name| drain(name, boundary).map { |event| [name, event] } }
+        end
+
+        def drain(name, boundary)
+          time_sorted(stream_scope(name)).between(boundary..boundary).to_a
+        end
+
+        def stream_scope(name)
+          name == SERIALIZED_GLOBAL_STREAM_NAME ? event_store.read : event_store.read.stream(name)
+        end
+
+        def group(rows)
+          rows
+            .group_by { |_, event| event.event_id }
+            .map { |_, list| [list.map(&:first).uniq, list.first.last] }
+            .sort_by { |_, event| [time_of(event), event.event_id] }
+            .reverse
+        end
+
+        def time_sorted(scope)
+          as_of? ? scope.as_of : scope.as_at
+        end
+
+        def time_of(event)
+          event.metadata.fetch(as_of? ? :valid_at : :timestamp)
+        end
+
+        def as_of?
+          sort == "as_of"
+        end
+
+        def cursor_time
+          @cursor_time ||= cursor && Time.iso8601(cursor)
+        end
+      end
+  end
+end

ruby_event_store-browser/lib/ruby_event_store/browser/urls.rb

@@ -39,6 +39,14 @@ def stream_page_url(stream_name, cursor, count)
         "#{stream_url(stream_name)}?#{query}"
       end
 
+      def swimlane_url(stream_names, sort = nil)
+        "#{app_url}/swimlane?#{swimlane_query(stream_names, sort)}"
+      end
+
+      def swimlane_more_url(stream_names, cursor, sort)
+        "#{app_url}/swimlane/more?#{swimlane_query(stream_names, sort, [["cursor", cursor]])}"
+      end
+
       def browser_js_url
         "#{app_url}/#{BROWSER_JS}"
       end
@@ -50,6 +58,15 @@ def browser_css_url
       def ==(other)
         self.class.eql?(other.class) && app_url.eql?(other.app_url)
       end
+
+      private
+
+      def swimlane_query(stream_names, sort, extra = [])
+        pairs = stream_names.map { |name| ["streams[]", name] }
+        pairs.concat(extra)
+        pairs << ["sort", sort] if sort
+        URI.encode_www_form(pairs)
+      end
     end
   end
 end

ruby_event_store-browser/lib/ruby_event_store/browser/views/streams/show.html.erb

@@ -15,9 +15,11 @@
       </ul>
     </div>
   </div>
-  <% if extension_links.any? -%>
+  <% swimlane_link = { label: "Swimlane view", url: urls.swimlane_url([stream_name]) } unless stream_name == SERIALIZED_GLOBAL_STREAM_NAME -%>
+  <% links = [swimlane_link, *extension_links].compact -%>
+  <% if links.any? -%>
     <div class="pt-3 flex gap-2">
-      <% extension_links.each do |link| -%>
+      <% links.each do |link| -%>
         <a class="inline-block text-center text-sm bg-red-700 text-gray-100 border border-red-700 rounded px-2 py-1" href="<%= link[:url] %>"><%= h(link[:label]) %></a>
       <% end -%>
     </div>

ruby_event_store-browser/lib/ruby_event_store/browser/views/swimlane/_rows.html.erb

@@ -0,0 +1,20 @@
+<% time_field = sort == "as_of" ? :valid_at : :timestamp -%>
+<% events.each_with_index do |(names, event), index| -%>
+  <% next_event = events[index + 1]&.last -%>
+  <% same_group = next_event && next_event.metadata[time_field].utc.iso8601(3) == event.metadata[time_field].utc.iso8601(3) -%>
+  <% cell = "align-middle px-2 py-1#{" border-b border-gray-300" unless same_group}" -%>
+  <tr>
+    <td class="<%= cell %> font-mono text-xs text-gray-500 whitespace-nowrap">
+      <% unless same_group -%>
+        <span data-swimlane-target="time" data-iso="<%= event.metadata[time_field].utc.iso8601(6) %>" title="UTC"><%= event.metadata[time_field].utc.strftime("%Y-%m-%dT%H:%M:%S.%3N") %></span>
+      <% end -%>
+    </td>
+    <% stream_names.each_with_index do |stream_name, column| -%>
+      <td class="<%= cell %><%= " border-l border-gray-200" if column.positive? %><%= " hover:bg-gray-100" if names.include?(stream_name) %>">
+        <% if names.include?(stream_name) -%>
+          <a href="<%= urls.event_url(event.event_id) %>" class="no-underline text-red-700 text-sm break-words"><%= h(event.event_type) %></a>
+        <% end -%>
+      </td>
+    <% end -%>
+  </tr>
+<% end -%>

ruby_event_store-browser/lib/ruby_event_store/browser/views/swimlane/show.html.erb

@@ -0,0 +1,50 @@
+<div class="py-8 container mx-auto" data-controller="swimlane swimlane-add" data-swimlane-more-url-value="<%= more_url %>">
+  <div class="flex justify-between items-center">
+    <h1 class="font-bold text-2xl">Comparing <%= h(stream_names.join(", ")) %></h1>
+    <button class="text-center text-sm border-red-700 text-red-700 border rounded px-2 py-1" data-action="swimlane-add#open">Add stream</button>
+  </div>
+  <div class="pt-3 flex items-center gap-2 text-sm">
+    <span class="text-gray-500">Sort by</span>
+    <% if sort == "as_of" -%>
+      <a class="no-underline text-gray-500 hover:text-red-700" href="<%= urls.swimlane_url(stream_names) %>">created at</a>
+      <span class="font-bold text-red-700">valid at</span>
+    <% else -%>
+      <span class="font-bold text-red-700">created at</span>
+      <a class="no-underline text-gray-500 hover:text-red-700" href="<%= urls.swimlane_url(stream_names, "as_of") %>">valid at</a>
+    <% end -%>
+  </div>
+
+  <div class="my-10 overflow-x-auto">
+    <table class="text-left">
+      <thead class="align-bottom leading-tight text-gray-500 uppercase text-xs">
+        <tr>
+          <th class="font-bold border-b border-gray-400 pb-2 px-2 whitespace-nowrap normal-case text-base text-gray-800">Time</th>
+          <% stream_names.each_with_index do |stream_name, column| -%>
+            <th class="font-bold border-b border-gray-400 pb-2 px-2 normal-case text-base text-gray-800<%= " border-l" if column.positive? %>">
+              <div class="flex justify-between items-center gap-2">
+                <span class="break-words"><%= h(stream_name) %></span>
+                <% if stream_names.size > 1 -%>
+                  <a class="text-gray-400 hover:text-red-700 no-underline text-xl leading-none" href="<%= urls.swimlane_url(stream_names - [stream_name], sort) %>" title="Remove from comparison">&times;</a>
+                <% end -%>
+              </div>
+            </th>
+          <% end -%>
+        </tr>
+      </thead>
+      <tbody data-swimlane-target="tbody">
+        <%= render("swimlane/_rows", urls: urls, stream_names: stream_names, events: events, sort: sort) %>
+      </tbody>
+    </table>
+  </div>
+
+  <dialog data-swimlane-add-target="dialog" class="backdrop:bg-gray-400/50 backdrop:backdrop-blur max-w-96 p-4 rounded-lg bg-white shadow w-full">
+    <button class="inset-0 fixed z-0" data-action="swimlane-add#close"></button>
+    <div class="isolate">
+      <form data-action="swimlane-add#go">
+        <div class="relative">
+          <input data-swimlane-add-target="input" class="rounded text-gray-800 cursor-pointer px-3 py-2 w-full appearance-none outline-none focus:ring-2 focus:ring-red-500 focus:ring-opacity-50" placeholder="Add stream to compare&hellip;" autofocus>
+        </div>
+      </form>
+    </div>
+  </dialog>
+</div>

ruby_event_store-browser/public/ruby_event_store_browser.js

@@ -92,3 +92,107 @@ application.register(
     }
   },
 )
+
+application.register(
+  "swimlane",
+  class extends Controller {
+    static targets = ["tbody", "time"]
+    static values = { moreUrl: String }
+
+    get storageKey() {
+      return "ruby_event_store_browser.timezone"
+    }
+
+    connect() {
+      this.onScroll = () => this.catchUp()
+      window.addEventListener("scroll", this.onScroll, { passive: true })
+      this.onZoneChange = (event) => {
+        if (event.target.matches('[data-timezone-target="select"]')) this.renderTimes(event.target.value)
+      }
+      document.addEventListener("change", this.onZoneChange)
+      this.renderTimes(this.zone())
+      this.catchUp()
+    }
+
+    disconnect() {
+      window.removeEventListener("scroll", this.onScroll)
+      document.removeEventListener("change", this.onZoneChange)
+    }
+
+    catchUp() {
+      if (!this.moreUrlValue) return
+      if (document.body.scrollHeight - (window.scrollY + window.innerHeight) > 200) return
+      this.loadMore()
+    }
+
+    loadMore() {
+      const url = this.moreUrlValue
+      if (!url) return
+      this.moreUrlValue = ""
+
+      fetch(url, { headers: { Accept: "application/json" } })
+        .then((response) => response.json())
+        .then(({ html, more_url }) => {
+          this.tbodyTarget.insertAdjacentHTML("beforeend", html)
+          this.moreUrlValue = more_url || ""
+          this.renderTimes(this.zone())
+          this.catchUp()
+        })
+    }
+
+    zone() {
+      const stored = localStorage.getItem(this.storageKey)
+      const detected = Intl.DateTimeFormat().resolvedOptions().timeZone
+      try {
+        Intl.DateTimeFormat("en-US", { timeZone: stored })
+        return stored || detected
+      } catch (_) {
+        return detected
+      }
+    }
+
+    renderTimes(tz) {
+      this.timeTargets.forEach((el) => {
+        el.textContent = this.format(el.dataset.iso, tz)
+        el.setAttribute("title", tz)
+      })
+    }
+
+    format(iso, tz) {
+      const parts = Object.fromEntries(
+        new Intl.DateTimeFormat("en-US", {
+          timeZone: tz, year: "numeric", month: "2-digit", day: "2-digit",
+          hour: "2-digit", minute: "2-digit", second: "2-digit",
+          fractionalSecondDigits: 3, hour12: false,
+        }).formatToParts(new Date(iso)).map((p) => [p.type, p.value])
+      )
+      return `${parts.year}-${parts.month}-${parts.day}T${parts.hour}:${parts.minute}:${parts.second}.${parts.fractionalSecond}`
+    }
+  },
+)
+
+application.register(
+  "swimlane-add",
+  class extends Controller {
+    static targets = ["dialog", "input"]
+
+    open(event) {
+      event?.preventDefault()
+      this.dialogTarget.showModal()
+      this.inputTarget.focus()
+    }
+
+    close() {
+      this.dialogTarget.close()
+    }
+
+    go(event) {
+      event.preventDefault()
+      const name = this.inputTarget.value
+      if (!name) return
+      const url = new URL(window.location.href)
+      url.searchParams.append("streams[]", name)
+      window.location = url.toString()
+    }
+  },
+)