Field LUTs + sync job + remove raw_data

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

@@ -2,6 +2,7 @@ module Api
   module Admin
     module V1
       class AdminController < Api::Admin::V1::ApplicationController
+        include HeartbeatFilterConcern
         before_action :can_write!, only: [ :user_convict ]
 
         def check
@@ -436,11 +437,14 @@ def user_heartbeats
             query = query.where("time <= ?", end_timestamp) if end_timestamp
           end
 
-          query = query.where(project: project) if project.present?
-          query = query.where(language: language) if language.present?
-          query = query.where(entity: entity) if entity.present?
-          query = query.where(editor: editor) if editor.present?
-          query = query.where(machine: machine) if machine.present?
+          filters = {
+            project: project,
+            language: language,
+            entity: entity,
+            editor: editor,
+            machine: machine
+          }.compact_blank
+          query = apply_heartbeat_filters(query, filters, user: user)
 
           total_count = query.count
 

app/controllers/api/summary_controller.rb

@@ -1,5 +1,6 @@
 module Api
   class SummaryController < ApplicationController
+    include HeartbeatFilterConcern
     skip_before_action :verify_authenticity_token
 
     def index
@@ -101,18 +102,21 @@ def determine_date_range(interval, range, from_date, to_date)
     end
 
     def filter_heartbeats(heartbeats, params)
-      heartbeats = heartbeats.where(project: params[:project]) if params[:project].present?
-      heartbeats = heartbeats.where(language: params[:language]) if params[:language].present?
-      heartbeats = heartbeats.where(editor: params[:editor]) if params[:editor].present?
-      heartbeats = heartbeats.where(operating_system: params[:operating_system]) if params[:operating_system].present?
-      heartbeats = heartbeats.where(machine: params[:machine]) if params[:machine].present?
-
+      user = nil
       if params[:user].present?
         user = User.find_by(slack_uid: params[:user])
         heartbeats = heartbeats.where(user_id: user.id) if user
       end
 
-      heartbeats
+      filters = {
+        project: params[:project],
+        language: params[:language],
+        editor: params[:editor],
+        operating_system: params[:operating_system],
+        machine: params[:machine]
+      }.compact_blank
+
+      apply_heartbeat_filters(heartbeats, filters, user: user)
     end
 
     def calculate_summary(heartbeats, date_range)

app/controllers/concerns/heartbeat_filter_concern.rb

@@ -0,0 +1,37 @@
+# frozen_string_literal: true
+
+# Case-insensitive filtering for heartbeat fields where display names differ from raw DB values.
+module HeartbeatFilterConcern
+  extend ActiveSupport::Concern
+
+  private
+
+  CASE_INSENSITIVE_FIELDS = %i[language editor operating_system].freeze
+
+  def apply_heartbeat_filter(scope, field, val, user: nil)
+    return scope if val.blank?
+    vals = Array(val)
+    CASE_INSENSITIVE_FIELDS.include?(field) ? case_insensitive_filter(scope, field, vals, user) : scope.where(field => vals)
+  end
+
+  def case_insensitive_filter(scope, field, vals, user)
+    raw = (user ? user.heartbeats : scope).distinct.pluck(field).compact_blank
+    matches = raw.select { |r| vals.any? { |v| v.casecmp?(r) || v.casecmp?(display_name(field, r)) } }
+    matches.any? ? scope.where(field => matches) : scope.none
+  end
+
+  def display_name(field, val)
+    h = ApplicationController.helpers
+    case field
+    when :language then WakatimeService.categorize_language(val) || val
+    when :editor then h.display_editor_name(val)
+    when :operating_system then h.display_os_name(val)
+    else val
+    end
+  end
+
+  def apply_heartbeat_filters(scope, filters, user: nil)
+    filters.each { |f, v| scope = apply_heartbeat_filter(scope, f, v, user: user) }
+    scope
+  end
+end

app/controllers/static_pages_controller.rb

@@ -201,7 +201,7 @@ def filterable_dashboard_data
           hb = if %i[operating_system editor].include?(f)
             hb.where(f => arr.flat_map { |v| [ v.downcase, v.capitalize ] }.uniq)
           elsif f == :language
-            raw = current_user.heartbeats.distinct.pluck(f).compact_blank.select { |l| arr.include?(l.categorize_language) }
+            raw = current_user.heartbeats.distinct.pluck(f).compact_blank.select { |l| arr.any? { |a| a.casecmp?(l.categorize_language) } }
             raw.any? ? hb.where(f => raw) : hb
           else
             hb.where(f => arr)

app/jobs/backfill_heartbeat_dimensions_job.rb

@@ -0,0 +1,97 @@
+class BackfillHeartbeatDimensionsJob < ApplicationJob
+  queue_as :latency_5m
+
+  include GoodJob::ActiveJobExtensions::Concurrency
+
+  good_job_control_concurrency_with(
+    key: -> { "backfill_heartbeat_dimensions_#{arguments.first}" },
+    total_limit: 1
+  )
+
+  BATCH_SIZE = 5_000
+
+  def perform(dimension, start_id: 0, end_id: nil)
+    spec = HeartbeatDimensionResolver::DIMENSIONS[dimension.to_sym]
+    unless spec
+      Rails.logger.error("Invalid dimension: #{dimension}")
+      return
+    end
+
+    end_id ||= Heartbeat.with_deleted.maximum(:id) || 0
+    current_id = start_id
+    processed = 0
+
+    while current_id <= end_id
+      batch_end = current_id + BATCH_SIZE
+      model = spec[:model].constantize
+
+      processed += if spec[:scope] == :global
+        backfill_global_dimension(model, spec[:value_attr], spec[:fk], spec[:lookup], current_id, batch_end)
+      else
+        backfill_user_scoped_dimension(model, spec[:value_attr], spec[:fk], current_id, batch_end)
+      end
+
+      current_id = batch_end
+      sleep(0.1)
+    end
+
+    Rails.logger.info("BackfillHeartbeatDimensionsJob: #{dimension} complete, processed #{processed} rows")
+  end
+
+  private
+
+  def backfill_global_dimension(model, string_column, fk_column, lookup_column, start_id, end_id)
+    heartbeats = Heartbeat.with_deleted
+                          .where(id: start_id...end_id)
+                          .where(fk_column => nil)
+                          .where.not(string_column => nil)
+
+    values = heartbeats.distinct.pluck(string_column).compact
+    return 0 if values.empty?
+
+    now = Time.current
+    rows = values.map { |v| { lookup_column => v, created_at: now, updated_at: now } }
+    model.upsert_all(rows, unique_by: lookup_column)
+
+    lookup_map = model.where(lookup_column => values).pluck(lookup_column, :id).to_h
+
+    updated = 0
+    lookup_map.each do |value, id|
+      updated += Heartbeat.with_deleted
+                          .where(id: start_id...end_id)
+                          .where(fk_column => nil)
+                          .where(string_column => value)
+                          .update_all(fk_column => id)
+    end
+    updated
+  end
+
+  def backfill_user_scoped_dimension(model, string_column, fk_column, start_id, end_id)
+    heartbeats = Heartbeat.with_deleted
+                          .where(id: start_id...end_id)
+                          .where(fk_column => nil)
+                          .where.not(string_column => nil)
+
+    user_value_pairs = heartbeats.distinct.pluck(:user_id, string_column).select { |u, v| u && v }
+    return 0 if user_value_pairs.empty?
+
+    now = Time.current
+    rows = user_value_pairs.map { |user_id, name| { user_id: user_id, name: name, created_at: now, updated_at: now } }
+    model.upsert_all(rows, unique_by: [ :user_id, :name ])
+
+    lookup_map = model.where([ "(user_id, name) IN (?)", user_value_pairs ])
+                      .pluck(:user_id, :name, :id)
+                      .each_with_object({}) { |(uid, name, id), h| h[[ uid, name ]] = id }
+
+    updated = 0
+    lookup_map.each do |(user_id, name), id|
+      updated += Heartbeat.with_deleted
+                          .where(id: start_id...end_id)
+                          .where(fk_column => nil)
+                          .where(user_id: user_id)
+                          .where(string_column => name)
+                          .update_all(fk_column => id)
+    end
+    updated
+  end
+end

app/jobs/migrate_user_from_hackatime_job.rb

@@ -70,20 +70,18 @@ def import_heartbeats
           cursorpos: heartbeat.cursor_position,
           project_root_count: heartbeat.project_root_count,
           is_write: heartbeat.is_write,
-          source_type: :wakapi_import,
-          raw_data: heartbeat.attributes.slice(*Heartbeat.indexed_attributes)
+          source_type: :wakapi_import
         }
-
         {
           **attrs,
           fields_hash: Heartbeat.generate_fields_hash(attrs)
         }
       end
 
-      # dedupe records by fields_hash
       records_to_upsert = records_to_upsert.group_by { |r| r[:fields_hash] }.map do |_, records|
         records.max_by { |r| r[:time] }
       end
+      records_to_upsert = Heartbeat.batch_resolve_dimensions(records_to_upsert)
 
       Heartbeat.upsert_all(records_to_upsert, unique_by: [ :fields_hash ])
     end

app/models/concerns/heartbeat_dimension_resolver.rb

@@ -0,0 +1,131 @@
+module HeartbeatDimensionResolver
+  extend ActiveSupport::Concern
+
+  DIMENSIONS = {
+    language:         { model: "Heartbeats::Language",        value_attr: :language,         fk: :language_id,         lookup: :name,  scope: :global },
+    category:         { model: "Heartbeats::Category",        value_attr: :category,         fk: :category_id,         lookup: :name,  scope: :global },
+    editor:           { model: "Heartbeats::Editor",          value_attr: :editor,           fk: :editor_id,           lookup: :name,  scope: :global },
+    operating_system: { model: "Heartbeats::OperatingSystem", value_attr: :operating_system, fk: :operating_system_id, lookup: :name,  scope: :global },
+    user_agent:       { model: "Heartbeats::UserAgent",       value_attr: :user_agent,       fk: :user_agent_id,       lookup: :value, scope: :global },
+    project:          { model: "Heartbeats::Project",         value_attr: :project,          fk: :project_id,          lookup: :name,  scope: :user },
+    branch:           { model: "Heartbeats::Branch",          value_attr: :branch,           fk: :branch_id,           lookup: :name,  scope: :user },
+    machine:          { model: "Heartbeats::Machine",         value_attr: :machine,          fk: :machine_id,          lookup: :name,  scope: :user }
+  }.freeze
+
+  included do
+    DIMENSIONS.each do |key, spec|
+      belongs_to :"heartbeat_#{key}",
+        class_name: spec[:model],
+        foreign_key: spec[:fk],
+        optional: true
+    end
+
+    before_save :resolve_dimension_ids, if: :should_resolve_dimensions?
+  end
+
+  private
+
+  def should_resolve_dimensions?
+    Flipper.enabled?(:heartbeat_dimension_dual_write)
+  end
+
+  def resolve_dimension_ids
+    DIMENSIONS.each_value do |spec|
+      next if self[spec[:fk]].present?
+
+      value = self[spec[:value_attr]]
+      next if value.blank?
+
+      model = spec[:model].constantize
+      resolved_id = if spec[:scope] == :global
+        model.resolve(value)&.id
+      else
+        model.resolve(user_id, value)&.id if user_id.present?
+      end
+
+      self[spec[:fk]] = resolved_id if resolved_id
+    end
+  end
+
+  class_methods do
+    def resolve_dimensions_for_attributes(attrs)
+      user_id = attrs[:user_id]
+
+      DIMENSIONS.each_value do |spec|
+        next if attrs[spec[:fk]].present?
+
+        value = attrs[spec[:value_attr]]
+        next if value.blank?
+
+        model = spec[:model].constantize
+        attrs[spec[:fk]] = if spec[:scope] == :global
+          model.resolve(value)&.id
+        else
+          model.resolve(user_id, value)&.id if user_id.present?
+        end
+      end
+
+      attrs
+    end
+
+    def batch_resolve_dimensions(records_attrs)
+      return records_attrs unless Flipper.enabled?(:heartbeat_dimension_dual_write)
+
+      global_specs = DIMENSIONS.select { |_, s| s[:scope] == :global }
+      user_specs = DIMENSIONS.select { |_, s| s[:scope] == :user }
+
+      global_maps = global_specs.transform_values do |spec|
+        values = records_attrs.filter_map { |r| r[spec[:value_attr]] }.uniq
+        batch_resolve_global(spec[:model].constantize, spec[:lookup], values)
+      end
+
+      user_maps = user_specs.transform_values do |spec|
+        pairs = records_attrs.filter_map { |r|
+          uid, val = r[:user_id], r[spec[:value_attr]]
+          [ uid, val ] if uid && val
+        }.uniq
+        batch_resolve_user_scoped(spec[:model].constantize, pairs)
+      end
+
+      records_attrs.map do |attrs|
+        attrs = attrs.dup
+
+        global_specs.each do |key, spec|
+          next if attrs[spec[:fk]].present?
+          attrs[spec[:fk]] = global_maps[key][attrs[spec[:value_attr]]] if attrs[spec[:value_attr]]
+        end
+
+        user_specs.each do |key, spec|
+          next if attrs[spec[:fk]].present?
+          attrs[spec[:fk]] = user_maps[key][[ attrs[:user_id], attrs[spec[:value_attr]] ]] if attrs[spec[:value_attr]] && attrs[:user_id]
+        end
+
+        attrs
+      end
+    end
+
+    private
+
+    def batch_resolve_global(model, column, values)
+      return {} if values.empty?
+
+      now = Time.current
+      rows = values.map { |v| { column => v, created_at: now, updated_at: now } }
+      model.upsert_all(rows, unique_by: column, returning: [ :id, column ])
+
+      model.where(column => values).pluck(column, :id).to_h
+    end
+
+    def batch_resolve_user_scoped(model, user_value_pairs)
+      return {} if user_value_pairs.empty?
+
+      now = Time.current
+      rows = user_value_pairs.map { |user_id, name| { user_id: user_id, name: name, created_at: now, updated_at: now } }
+      model.upsert_all(rows, unique_by: [ :user_id, :name ], returning: [ :id, :user_id, :name ])
+
+      model.where([ "(user_id, name) IN (?)", user_value_pairs ])
+           .pluck(:user_id, :name, :id)
+           .each_with_object({}) { |(uid, name, id), h| h[[ uid, name ]] = id }
+    end
+  end
+end

app/models/heartbeat.rb

@@ -1,10 +1,10 @@
 class Heartbeat < ApplicationRecord
   before_save :set_fields_hash!
-  before_save :set_raw_data!
 
   include Heartbeatable
   include TimeRangeFilterable
   include PublicActivity::Common
+  include HeartbeatDimensionResolver
 
   time_range_filterable_field :time
 
@@ -112,10 +112,6 @@ def self.indexed_attributes
     %w[user_id branch category dependencies editor entity language machine operating_system project type user_agent line_additions line_deletions lineno lines cursorpos project_root_count time is_write]
   end
 
-  def set_raw_data!
-    self.raw_data ||= self.attributes.slice(*self.class.indexed_attributes)
-  end
-
   def soft_delete
     update_column(:deleted_at, Time.current)
   end

app/models/heartbeats/branch.rb

@@ -0,0 +1,5 @@
+class Heartbeats::Branch < Heartbeats::UserScopedLookupBase
+  self.table_name = "heartbeat_branches"
+
+  has_many :heartbeats, foreign_key: :branch_id, inverse_of: :heartbeat_branch
+end

app/models/heartbeats/category.rb

@@ -0,0 +1,5 @@
+class Heartbeats::Category < Heartbeats::LookupBase
+  self.table_name = "heartbeat_categories"
+
+  has_many :heartbeats, foreign_key: :category_id, inverse_of: :heartbeat_category
+end