[HEAVILY WIP] Migrate to ClickHouse

.github/workflows/ci.yml

@@ -149,6 +149,17 @@ jobs:
           POSTGRES_USER: postgres
           POSTGRES_PASSWORD: postgres
           POSTGRES_DB: app_test
+      clickhouse:
+        image: clickhouse/clickhouse-server:26.6.1.1193
+        ports:
+          - 8123:8123
+        options: >-
+          --health-cmd "clickhouse-client --query 'SELECT 1'"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+        env:
+          CLICKHOUSE_DB: hackatime_test
 
     steps:
       - name: Checkout code
@@ -167,6 +178,11 @@ jobs:
           PGHOST: localhost
           PGUSER: postgres
           PGPASSWORD: postgres
+          CLICKHOUSE_HOST: localhost
+          CLICKHOUSE_PORT: 8123
+          CLICKHOUSE_TEST_DATABASE: hackatime_test
+          CLICKHOUSE_USER: default
+          CLICKHOUSE_PASSWORD: ""
         run: |
           bin/rails db:create RAILS_ENV=test
           bin/rails db:schema:load RAILS_ENV=test
@@ -179,6 +195,11 @@ jobs:
           PGHOST: localhost
           PGUSER: postgres
           PGPASSWORD: postgres
+          CLICKHOUSE_HOST: localhost
+          CLICKHOUSE_PORT: 8123
+          CLICKHOUSE_TEST_DATABASE: hackatime_test
+          CLICKHOUSE_USER: default
+          CLICKHOUSE_PASSWORD: ""
         run: |
           bin/rails rswag:specs:swaggerize
           git diff --exit-code swagger/
@@ -196,6 +217,17 @@ jobs:
           POSTGRES_USER: postgres
           POSTGRES_PASSWORD: postgres
           POSTGRES_DB: app_test
+      clickhouse:
+        image: clickhouse/clickhouse-server:26.6.1.1193
+        ports:
+          - 8123:8123
+        options: >-
+          --health-cmd "clickhouse-client --query 'SELECT 1'"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+        env:
+          CLICKHOUSE_DB: hackatime_test
 
     steps:
       - name: Checkout code
@@ -236,6 +268,11 @@ jobs:
           PGHOST: localhost
           PGUSER: postgres
           PGPASSWORD: postgres
+          CLICKHOUSE_HOST: localhost
+          CLICKHOUSE_PORT: 8123
+          CLICKHOUSE_TEST_DATABASE: hackatime_test
+          CLICKHOUSE_USER: default
+          CLICKHOUSE_PASSWORD: ""
           CHROME_BIN: ${{ steps.setup-chrome.outputs.chrome-path }}
           CHROMEDRIVER_BIN: ${{ steps.setup-chrome.outputs.chromedriver-path }}
         run: |

Gemfile

@@ -9,6 +9,8 @@ gem "propshaft"
 gem "sqlite3", ">= 2.1"
 # Use PostgreSQL as the database for Wakatime
 gem "pg"
+# ClickHouse ActiveRecord adapter - heartbeat reads + writes run on ClickHouse
+gem "clickhouse-activerecord"
 # Use the Puma web server [https://github.com/puma/puma]
 gem "puma", ">= 5.0"
 # Use JavaScript with ESM import maps [https://github.com/rails/importmap-rails]

Gemfile.lock

@@ -131,6 +131,9 @@ GEM
     childprocess (5.1.0)
       logger (~> 1.5)
     chunky_png (1.4.0)
+    clickhouse-activerecord (1.6.7)
+      activerecord (>= 7.1, < 9.0)
+      bundler (>= 1.13.4)
     cloudflare-rails (7.0.0)
       actionpack (>= 7.2.0, < 8.2.0)
       activesupport (>= 7.2.0, < 8.2.0)
@@ -594,6 +597,7 @@ DEPENDENCIES
   brakeman
   bullet
   capybara
+  clickhouse-activerecord
   cloudflare-rails
   countries
   debug

app/controllers/admin/account_merger_controller.rb

@@ -107,9 +107,49 @@
       results << "user ##{newer_user.id} deleted"
     end
 
+    mirror_merge_to_clickhouse(older_user:, newer_user:)
+
     results.join(", ")
   end
 
+  # update_all(user_id:) bypasses the mirror callback, so replay the move in
+  # ClickHouse server-side: copy the live rows under the older user, then
+  # tombstone everything left under the newer user.
+  def mirror_merge_to_clickhouse(older_user:, newer_user:)
+    return unless Clickhouse::HeartbeatMirror.enabled?
+
+    connection = Clickhouse::Heartbeat.connection
+    table = connection.quote_table_name(Clickhouse::Heartbeat.table_name)
+    columns = Clickhouse::Heartbeat.column_names
+    column_list = columns.map { |column| connection.quote_column_name(column) }.join(", ")
+    version = (Time.current.to_f * 1_000_000).round
+
+    moved_exprs = columns.map do |column|
+      column == "user_id" ? "#{older_user.id.to_i} AS user_id" : connection.quote_column_name(column)
+    end
+    connection.execute(<<~SQL.squish)
+      INSERT INTO #{table} (#{column_list})
+      SELECT #{moved_exprs.join(", ")} FROM #{table} FINAL
+      WHERE user_id = #{newer_user.id.to_i} AND deleted_at IS NULL
+    SQL
+
+    tombstone_exprs = columns.map do |column|
+      case column
+      when "deleted_at" then "now64(6) AS deleted_at"
+      when "version" then "#{version.to_i} AS version"
+      else connection.quote_column_name(column)
+      end
+    end
+    connection.execute(<<~SQL.squish)
+      INSERT INTO #{table} (#{column_list})
+      SELECT #{tombstone_exprs.join(", ")} FROM #{table} FINAL
+      WHERE user_id = #{newer_user.id.to_i} AND deleted_at IS NULL
+    SQL
+  rescue => e
+    Rails.logger.error("ClickHouse mirror of account merge failed (Postgres merge already committed): #{e.class}: #{e.message}")
+    report_error(e, message: "ClickHouse mirror of account merge failed for users #{older_user.id} <- #{newer_user.id}")
+  end
+
   DELETABLE_TABLES = %w[heartbeat_import_sources wakatime_mirrors project_labels].freeze
 
   def transfer_api_keys(older_user:, newer_user:)

app/controllers/api/admin/v1/admin_controller.rb

@@ -38,76 +38,83 @@ def visualization_quantized
           quantized_query = <<-SQL
             WITH base_heartbeats AS (
                 SELECT
-                    "time",
+                    time,
                     lineno,
                     cursorpos,
-                    date_trunc('day', to_timestamp("time")) as day_start
-                FROM heartbeats
+                    toStartOfDay(toDateTime(time)) as day_start
+                FROM heartbeats FINAL
                 WHERE user_id = ?
                 AND deleted_at IS NULL
-                AND "time" >= ? AND "time" <= ?
+                AND time >= ? AND time <= ?
                 LIMIT 1000000
             ),
             daily_stats AS (
                 SELECT
                     *,
-                    GREATEST(1, MAX(lineno) OVER (PARTITION BY day_start)) as max_lineno,
-                    GREATEST(1, MAX(cursorpos) OVER (PARTITION BY day_start)) as max_cursorpos
+                    greatest(1, ifNull(MAX(lineno) OVER (PARTITION BY day_start), 1)) as max_lineno,
+                    greatest(1, ifNull(MAX(cursorpos) OVER (PARTITION BY day_start), 1)) as max_cursorpos
                 FROM base_heartbeats
             ),
             quantized_heartbeats AS (
                 SELECT
                     *,
-                    ROUND(2 + (("time" - extract(epoch from day_start)) / 86400) * (396)) as qx,
-                    ROUND(2 + (1 - CAST(lineno AS decimal) / max_lineno) * (96)) as qy_lineno,
-                    ROUND(2 + (1 - CAST(cursorpos AS decimal) / max_cursorpos) * (96)) as qy_cursorpos
+                    round(2 + ((time - toUnixTimestamp(day_start)) / 86400) * (396)) as qx,
+                    round(2 + (1 - toFloat64(lineno) / max_lineno) * (96)) as qy_lineno,
+                    round(2 + (1 - toFloat64(cursorpos) / max_cursorpos) * (96)) as qy_cursorpos
                 FROM daily_stats
             )
-            SELECT "time", lineno, cursorpos
-            FROM (
-                SELECT DISTINCT ON (day_start, qx, qy_lineno) "time", lineno, cursorpos
-                FROM quantized_heartbeats
-                WHERE lineno IS NOT NULL
-                ORDER BY day_start, qx, qy_lineno, "time" ASC
-            ) AS lineno_pixels
-            UNION
-            SELECT "time", lineno, cursorpos
-            FROM (
-                SELECT DISTINCT ON (day_start, qx, qy_cursorpos) "time", lineno, cursorpos
-                FROM quantized_heartbeats
-                WHERE cursorpos IS NOT NULL
-                ORDER BY day_start, qx, qy_cursorpos, "time" ASC
-            ) AS cursorpos_pixels
-            UNION
-            SELECT "time", lineno, cursorpos
-            FROM (
-                SELECT DISTINCT ON (day_start, qx) "time", lineno, cursorpos
-                FROM quantized_heartbeats
-                WHERE lineno IS NULL AND cursorpos IS NULL
-                ORDER BY day_start, qx, "time" ASC
-            ) AS null_pixels
-            ORDER BY "time" ASC
+            SELECT time, lineno, cursorpos FROM (
+              SELECT time, lineno, cursorpos
+              FROM (
+                  SELECT time, lineno, cursorpos
+                  FROM quantized_heartbeats
+                  WHERE lineno IS NOT NULL
+                  ORDER BY day_start, qx, qy_lineno, time ASC
+                  LIMIT 1 BY day_start, qx, qy_lineno
+              ) AS lineno_pixels
+              UNION DISTINCT
+              SELECT time, lineno, cursorpos
+              FROM (
+                  SELECT time, lineno, cursorpos
+                  FROM quantized_heartbeats
+                  WHERE cursorpos IS NOT NULL
+                  ORDER BY day_start, qx, qy_cursorpos, time ASC
+                  LIMIT 1 BY day_start, qx, qy_cursorpos
+              ) AS cursorpos_pixels
+              UNION DISTINCT
+              SELECT time, lineno, cursorpos
+              FROM (
+                  SELECT time, lineno, cursorpos
+                  FROM quantized_heartbeats
+                  WHERE lineno IS NULL AND cursorpos IS NULL
+                  ORDER BY day_start, qx, time ASC
+                  LIMIT 1 BY day_start, qx
+              ) AS null_pixels
+            ) AS pixels
+            ORDER BY time ASC
           SQL
 
           daily_totals_query = <<-SQL
             WITH heartbeats_with_gaps AS (
               SELECT
-                date_trunc('day', to_timestamp("time"))::date as day,
-                "time" - LAG("time", 1, "time") OVER (PARTITION BY date_trunc('day', to_timestamp("time")) ORDER BY "time", id) as gap
-              FROM heartbeats
+                toDate(toDateTime(time)) as day,
+                time - lagInFrame(time, 1, time) OVER (
+                  PARTITION BY toStartOfDay(toDateTime(time)) ORDER BY time, fields_hash
+                  ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+                ) as gap
+              FROM heartbeats FINAL
               WHERE user_id = ? AND deleted_at IS NULL AND time >= ? AND time <= ?
             )
             SELECT day, SUM(LEAST(gap, 120)) as total_seconds
             FROM heartbeats_with_gaps
-            WHERE gap IS NOT NULL
             GROUP BY day
           SQL
 
-          conn = ActiveRecord::Base.connection
-          quantized_result = conn.execute(ActiveRecord::Base.sanitize_sql([ quantized_query, user.id, start_epoch, end_epoch ]))
-          daily_totals_result = conn.execute(ActiveRecord::Base.sanitize_sql([ daily_totals_query, user.id, start_epoch, end_epoch ]))
+          conn = Clickhouse::Heartbeat.connection
+          quantized_result = conn.select_all(Clickhouse::Heartbeat.sanitize_sql([ quantized_query, user.id, start_epoch, end_epoch ]))
+          daily_totals_result = conn.select_all(Clickhouse::Heartbeat.sanitize_sql([ daily_totals_query, user.id, start_epoch, end_epoch ]))
 
-          daily_totals = daily_totals_result.each_with_object({}) { |row, h| h[row["day"]] = row["total_seconds"] }
+          daily_totals = daily_totals_result.each_with_object({}) { |row, h| h[row["day"].to_date] = row["total_seconds"].to_f.round }
 
           points_by_day = quantized_result.each_with_object({}) do |row, hash|
             day = Time.at(row["time"]).to_date
@@ -130,8 +137,8 @@ def alt_candidates
             SELECT
                 r1.user_id AS user_a_id,
                 r2.user_id AS user_b_id,
-                r1.machine,
-                r1.ip_address,
+                r1.machine AS machine,
+                r1.ip_address AS ip_address,
                 r1.first_seen as user_a_first_seen_on_combo,
                 r1.last_seen as user_a_last_seen_on_combo,
                 r2.first_seen as user_b_first_seen_on_combo,
@@ -144,14 +151,13 @@ def alt_candidates
                         ip_address,
                         MIN(time) as first_seen,
                         MAX(time) as last_seen
-                    FROM heartbeats
+                    FROM heartbeats FINAL
                     WHERE
-                        user_id IS NOT NULL
-                        AND machine IS NOT NULL
+                        machine IS NOT NULL
                         AND ip_address IS NOT NULL
                         AND deleted_at IS NULL
                         AND time >= ?
-                    GROUP BY 1, 2, 3
+                    GROUP BY user_id, machine, ip_address
                 ) r1
             JOIN
                 (
@@ -161,22 +167,21 @@ def alt_candidates
                         ip_address,
                         MIN(time) as first_seen,
                         MAX(time) as last_seen
-                    FROM heartbeats
+                    FROM heartbeats FINAL
                     WHERE
-                        user_id IS NOT NULL
-                        AND machine IS NOT NULL
+                        machine IS NOT NULL
                         AND ip_address IS NOT NULL
                         AND deleted_at IS NULL
                         AND time >= ?
-                    GROUP BY 1, 2, 3
+                    GROUP BY user_id, machine, ip_address
                 ) r2 ON r1.machine = r2.machine AND r1.ip_address = r2.ip_address
             WHERE
                 r1.user_id < r2.user_id
             LIMIT 5000
           SQL
 
-          result = ActiveRecord::Base.connection.exec_query(
-            ActiveRecord::Base.sanitize_sql([ query, cutoff, cutoff ])
+          result = Clickhouse::Heartbeat.connection.select_all(
+            Clickhouse::Heartbeat.sanitize_sql([ query, cutoff, cutoff ])
           )
 
           render json: { candidates: result.to_a }
@@ -188,7 +193,7 @@ def active_users
           return render_error("invalid since parameter") if since_ts < 0
 
           since_ts = [ since_ts, 90.days.ago.to_i ].max
-          render json: { user_ids: Heartbeat.where("time >= ?", since_ts).distinct.limit(50_000).pluck(:user_id) }
+          render json: { user_ids: Clickhouse::Heartbeat.where("time >= ?", since_ts).distinct.limit(50_000).pluck(:user_id) }
         end
 
         def audit_logs_counts
@@ -217,18 +222,19 @@ def heartbeats_by_user_agent_segment
           user_id = params[:user_id].presence
 
           escaped = segment.gsub(/[\\%_]/) { |c| "\\#{c}" }
-          query = Heartbeat.where("user_agent ILIKE ?", "%#{escaped}%")
+          query = Clickhouse::Heartbeat.where("user_agent ILIKE ?", "%#{escaped}%")
           query = query.where(user_id: user_id) if user_id
           query = apply_time_range(query) or return
 
           if ActiveModel::Type::Boolean.new.cast(params[:count_only])
             return render json: { segment: segment, total_count: query.limit(nil).count }
           end
 
-          heartbeats = query.order(time: :desc).limit(limit + 1).offset(offset).to_a
+          heartbeats = query.order(time: :desc, fields_hash: :desc).limit(limit + 1).offset(offset).to_a
           has_more = heartbeats.size > limit
           heartbeats = heartbeats.first(limit)
 
+          source_types = Heartbeat.source_types.invert
           render json: {
             segment: segment,
             limit: limit,
@@ -252,7 +258,7 @@ def heartbeats_by_user_agent_segment
                 lineno: hb.lineno,
                 cursorpos: hb.cursorpos,
                 lines: hb.lines,
-                source_type: hb.source_type
+                source_type: source_types[hb.source_type] || hb.source_type
               }
             },
             has_more: has_more

app/controllers/api/admin/v1/trust_level_audit_logs_controller.rb

@@ -10,7 +10,9 @@ class TrustLevelAuditLogsController < Api::Admin::V1::ApplicationController
         }.freeze
 
         def index
-          audit_logs = TrustLevelAuditLog.includes(:user, :changed_by).recent.limit(250)
+          audit_logs = TrustLevelAuditLog
+            .includes(user: :email_addresses, changed_by: :email_addresses)
+            .recent.limit(250)
 
           if params[:user_id].present?
             user = User.find_by(id: params[:user_id])

app/controllers/api/hackatime/v1/hackatime_controller.rb

@@ -34,7 +34,7 @@ def push_heartbeats
 
   def status_bar_today
     Time.use_zone(@user.timezone) do
-      total_seconds = @user.heartbeats.today.duration_seconds
+      total_seconds = Clickhouse::Heartbeat.for_user(@user).today.duration_seconds
       result = {
         data: {
           grand_total: {
@@ -69,7 +69,7 @@ def stats_last_7_days
       start_time = (Time.current - 7.days).beginning_of_day
       end_time = Time.current.end_of_day
 
-      heartbeats = @user.heartbeats.where(time: start_time.to_i..end_time.to_i)
+      heartbeats = Clickhouse::Heartbeat.for_user(@user).where(time: start_time.to_i..end_time.to_i)
       total_seconds = heartbeats.duration_seconds.to_i
       days_covered = heartbeats.pluck(:time).map { |ts| Time.at(ts).in_time_zone(@user.timezone).to_date }.uniq.length
       daily_average = days_covered > 0 ? (total_seconds.to_f / days_covered).round(1) : 0