Implement publishing progress to kafka

Author: nygrenh

app/background_tasks/kafka_batch_update_points_task.rb

@@ -1,21 +1,50 @@
 # frozen_string_literal: true
 
-require "kafka"
-
+require 'kafka'
 
 class KafkaBatchUpdatePointsTask
   def initialize
     seed_brokers = SiteSetting.value('kafka_seed_brokers')
-    @kafka = seed_brokers && Kafka.new(seed_brokers, client_id: "tmc-server")
+    @service_id = SiteSetting.value('kafka_service_id')
+    @kafka = seed_brokers && Kafka.new(seed_brokers, client_id: 'tmc-server')
   end
 
   def run
-    return unless @kafka
+    return unless @kafka && @service_id
     producer = @kafka.producer
-    task = KafkaBatchUpdatePoints.first
-    course = task.course
-    puts "Batch publishing points for course #{course.name}."
-
+    KafkaBatchUpdatePoints.all.each do |task|
+      course = task.course
+      Rails.logger.info("Batch publishing points for course #{course.name} with moocfi id: #{course.moocfi_id}.")
+      if !course.moocfi_id
+        Rails.logger.error 'Cannot publish points because moocfi id is not specified'
+        next
+      end
+      points_per_user = AwardedPoint.count_per_user_in_course_with_sheet(course, course.gdocs_sheets)
+      Rails.logger.info("Found points for #{parts.keys.length} users")
+      available_points = AvailablePoint.course_sheet_points(course, parts)
+      points_per_user.each do |username, points_by_group|
+        user = User.find_by(login: username)
+        Rails.logger.info("Publishing points for user #{user.id}")
+        progress = points_by_group.map do |group_name, awareded_points|
+          max_points = available_points[group_name] || 0
+          {
+            group: group_name,
+            n_points: awareded_points,
+            max_points: max_points,
+            progress: (awareded_points / max_points.to_f).floor(2)
+          }
+        end
+        message = {
+          timestamp: Time.zone.now.iso8601,
+          user_id: user.id,
+          course_id: course.moocfi_id,
+          service_id: @service_id,
+          progress: progress,
+          message_format_version: 1
+        }
+        producer.deliver_message(message, topic: 'user-course-progress')
+      end
+    end
   end
 
   def wait_delay

db/migrate/20191101092508_add_moocfi_id_to_course.rb

@@ -0,0 +1,5 @@
+class AddMoocfiIdToCourse < ActiveRecord::Migration
+  def change
+    add_column :courses, :moocfi_id, :string
+  end
+end

db/schema.rb

@@ -11,7 +11,7 @@
 #
 # It's strongly recommended that you check this file into your version control system.
 
-ActiveRecord::Schema.define(version: 20200112230122) do
+ActiveRecord::Schema.define(version: 20200112230124) do
   # These are extensions that must be enabled in order to support this database
   enable_extension "plpgsql"