Refactor KafkaBatchUpdatePoints methods to enhance error handling, improve submission filtering, and aggregate results for point resending operations

Author: nygrenh

app/models/kafka_batch_update_points.rb

@@ -3,103 +3,143 @@
 class KafkaBatchUpdatePoints < ApplicationRecord
   belongs_to :course
 
-  # Optionally restrict to submissions within `interval` (ActiveSupport::Duration)
-  # e.g. send_points_again_for_user_and_course(course_id, user_id, interval: 2.weeks)
+  # Re-enqueue user progress and points updates for a given (course, user) pair.
+  #
+  # Only considers exercises the user has submitted in that course.
+  # Optionally limits to submissions within `interval` (ActiveSupport::Duration).
+  #
+  # Returns:
+  #   { status: :ok, course_id:, user_id:, progress_enqueued:, user_points_enqueued: }
+  #   or { status: :skip_<reason>, course_id:, user_id: }
   def self.send_points_again_for_user_and_course(course_id, user_id, interval: nil)
     if interval && !interval.is_a?(ActiveSupport::Duration)
-      raise ArgumentError,
-            'Invalid interval: expected an ActiveSupport::Duration (e.g. 1.week, 3.days). ' \
-            'Call like: KafkaBatchUpdatePoints.send_points_again_for_user_and_course(42, 7, interval: 1.week)'
+      raise ArgumentError, 'Invalid interval: expected an ActiveSupport::Duration (e.g. 1.week, 3.days)'
     end
 
-    course = Course.find_by!(id: course_id)
+    course = Course.find_by(id: course_id)
+    return { status: :skip_course_missing, course_id: course_id, user_id: user_id } if course.nil?
+    return { status: :skip_no_moocfi_id, course_id: course.id, user_id: user_id } if course.moocfi_id.blank?
 
-    # Skip if course doesn't exist or has no moocfi_id
-    if course.moocfi_id.nil? || course.moocfi_id.blank?
-      puts "⚠️  Skipping course_id=#{course_id} (missing or empty moocfi_id)"
-      return
-    end
-
-    submissions_scope = Submission
-                          .where(course_id: course.id, user_id: user_id)
-                          .where.not(exercise_name: [nil, ''])
-    submissions_scope = submissions_scope.where(created_at: (Time.current - interval)..Time.current) if interval
-
-    submitted_names = submissions_scope.distinct.pluck(:exercise_name)
-    if submitted_names.empty?
-      puts "ℹ️  No submissions found for user_id=#{user_id} in course_id=#{course.id}"
-      return
-    end
+    scope = Submission.where(course_id: course.id, user_id: user_id).where.not(exercise_name: [nil, ''])
+    scope = scope.where(created_at: (Time.current - interval)..Time.current) if interval
 
-    exercise_ids = Exercise
-                     .where(course_id: course.id, name: submitted_names)
-                     .distinct
-                     .pluck(:id)
+    names = scope.distinct.pluck(:exercise_name)
+    return { status: :skip_no_submissions, course_id: course.id, user_id: user_id } if names.empty?
 
-    if exercise_ids.empty?
-      puts "⚠️  No exercises matched submission names for user_id=#{user_id} in course_id=#{course.id}"
-      return
-    end
+    exercise_ids = Exercise.where(course_id: course.id, name: names).distinct.pluck(:id)
+    return { status: :skip_no_matching_exercises, course_id: course.id, user_id: user_id } if exercise_ids.empty?
 
     transaction do
       create!(course_id: course.id, user_id: user_id, realtime: false, task_type: 'user_progress')
-
       exercise_ids.each do |exercise_id|
         create!(course_id: course.id, user_id: user_id, exercise_id: exercise_id, realtime: false, task_type: 'user_points')
       end
     end
 
-    puts "✅  Resent points for user_id=#{user_id}, course_id=#{course.id} (#{exercise_ids.size} exercises)"
+    { status: :ok, course_id: course.id, user_id: user_id, progress_enqueued: 1, user_points_enqueued: exercise_ids.size }
   end
 
+  # Re-enqueue points for all courses a user has submissions in.
+  #
+  # Accepts optional `interval` (ActiveSupport::Duration) to restrict submissions.
+  # Returns aggregate counts by status and total enqueued items.
   def self.send_points_again_for_user_and_all_courses(user_id, interval: nil)
     if interval && !interval.is_a?(ActiveSupport::Duration)
-      raise ArgumentError,
-            'Invalid interval: expected an ActiveSupport::Duration (e.g. 1.week, 3.days). ' \
-            'Call like: KafkaBatchUpdatePoints.send_points_again_for_user_and_all_courses(7, interval: 3.days)'
+      raise ArgumentError, 'Invalid interval: expected an ActiveSupport::Duration (e.g. 1.week, 3.days)'
     end
 
-    course_ids = Submission.where(user_id: user_id).distinct.pluck(:course_id)
-    puts "🔁  Processing #{course_ids.size} courses for user_id=#{user_id}..."
-
-    course_ids.each do |course_id|
-      send_points_again_for_user_and_course(course_id, user_id, interval: interval)
+    totals = {
+      ok: 0, skip_course_missing: 0, skip_no_moocfi_id: 0, skip_no_submissions: 0, skip_no_matching_exercises: 0,
+      progress_enqueued: 0, user_points_enqueued: 0, processed_courses: 0
+    }
+
+    Submission.where(user_id: user_id).distinct.pluck(:course_id).each do |course_id|
+      r = send_points_again_for_user_and_course(course_id, user_id, interval: interval)
+      totals[:processed_courses] += 1
+      case r[:status]
+      when :ok
+        totals[:ok] += 1
+        totals[:progress_enqueued] += r[:progress_enqueued]
+        totals[:user_points_enqueued] += r[:user_points_enqueued]
+      else
+        totals[r[:status]] += 1 if totals.key?(r[:status])
+      end
     end
+
+    totals
   end
 
-  # Requires an ActiveSupport::Duration, e.g. 1.week, 3.days, etc.
-  def self.resend_points_for_recent_submissions(duration)
+  # Re-enqueue points for all (course, user) pairs with submissions in the last `duration`.
+  #
+  # Prints a per-batch summary for each batch (`batch_size` pairs per batch),
+  # then prints overall totals at the end.
+  # Wraps the whole operation in one transaction; if it aborts, prints a clear message.
+  # Returns overall totals hash.
+  def self.resend_points_for_recent_submissions(duration, batch_size: 1000)
     unless duration.is_a?(ActiveSupport::Duration)
-      raise ArgumentError,
-            'Invalid argument: expected an ActiveSupport::Duration (e.g. 1.week, 3.days). ' \
-            'Call it like: KafkaBatchUpdatePoints.resend_points_for_recent_submissions(1.week)'
+      raise ArgumentError, 'Invalid argument: expected an ActiveSupport::Duration (e.g. 1.week, 3.days)'
     end
 
-    pairs = Submission
-              .where(created_at: (Time.current - duration)..Time.current)
-              .select(:course_id, :user_id)
-              .distinct
-              .pluck(:course_id, :user_id)
-
-    puts "🔍  Found #{pairs.size} (course_id, user_id) pairs in the last #{duration.inspect}"
-
-    processed = 0
-    skipped = 0
+    pairs = Submission.where(created_at: (Time.current - duration)..Time.current)
+                      .select(:course_id, :user_id)
+                      .distinct
+                      .pluck(:course_id, :user_id)
+
+    overall = {
+      ok: 0, skip_course_missing: 0, skip_no_moocfi_id: 0, skip_no_submissions: 0, skip_no_matching_exercises: 0,
+      progress_enqueued: 0, user_points_enqueued: 0, total_pairs: pairs.size
+    }
+
+    begin
+      transaction do
+        pairs.each_slice(batch_size).with_index(1) do |batch_pairs, batch_idx|
+          batch = {
+            ok: 0, skip_course_missing: 0, skip_no_moocfi_id: 0, skip_no_submissions: 0, skip_no_matching_exercises: 0,
+            progress_enqueued: 0, user_points_enqueued: 0, processed_in_batch: batch_pairs.size
+          }
+
+          batch_pairs.each do |course_id, user_id|
+            r = send_points_again_for_user_and_course(course_id, user_id, interval: duration)
+            case r[:status]
+            when :ok
+              batch[:ok] += 1
+              batch[:progress_enqueued] += r[:progress_enqueued]
+              batch[:user_points_enqueued] += r[:user_points_enqueued]
+            else
+              batch[r[:status]] += 1 if batch.key?(r[:status])
+            end
+          end
 
-    transaction do
-      pairs.each do |course_id, user_id|
-          before = KafkaBatchUpdatePoints.count
-          send_points_again_for_user_and_course(course_id, user_id, interval: duration)
-          after = KafkaBatchUpdatePoints.count
-
-          if after == before
-            skipped += 1
-          else
-            processed += 1
+          # fold batch into overall
+          overall.keys.each do |k|
+            next if k == :total_pairs
+            overall[k] += batch[k] if batch.key?(k)
           end
+
+          skipped = batch.values_at(:skip_course_missing, :skip_no_moocfi_id, :skip_no_submissions, :skip_no_matching_exercises).sum
+          puts(
+            "Batch #{batch_idx} (size=#{batch[:processed_in_batch]}): " \
+            "ok=#{batch[:ok]}, skipped=#{skipped} " \
+            "(missing_course=#{batch[:skip_course_missing]}, no_moocfi=#{batch[:skip_no_moocfi_id]}, " \
+            "no_submissions=#{batch[:skip_no_submissions]}, no_match=#{batch[:skip_no_matching_exercises]}), " \
+            "enqueued(progress=#{batch[:progress_enqueued]}, user_points=#{batch[:user_points_enqueued]})"
+          )
         end
+      end
+
+      # Only printed if transaction committed
+      total_skipped = overall.values_at(:skip_course_missing, :skip_no_moocfi_id, :skip_no_submissions, :skip_no_matching_exercises).sum
+      puts(
+        "TOTALS: pairs=#{overall[:total_pairs]}, ok=#{overall[:ok]}, skipped=#{total_skipped} " \
+        "(missing_course=#{overall[:skip_course_missing]}, no_moocfi=#{overall[:skip_no_moocfi_id]}, " \
+        "no_submissions=#{overall[:skip_no_submissions]}, no_match=#{overall[:skip_no_matching_exercises]}), " \
+        "enqueued(progress=#{overall[:progress_enqueued]}, user_points=#{overall[:user_points_enqueued]})"
+      )
+    rescue => e
+      puts "⛔ Transaction aborted in resend_points_for_recent_submissions: #{e.class}: #{e.message}"
+      raise
     end
 
-    puts "✅  Done! Processed #{processed} pairs, skipped #{skipped} (#{pairs.size} total)"
+    overall
   end
 end