135 lines
3.7 KiB
Ruby
135 lines
3.7 KiB
Ruby
# frozen_string_literal: true
|
|
|
|
module Analytics
|
|
class Ingest
|
|
MAX_BATCH = 80
|
|
RATE_LIMIT_PER_MINUTE = 90
|
|
|
|
Result = Struct.new(:accepted, :rejected, :rate_limited, keyword_init: true)
|
|
|
|
def initialize(events:, remote_ip:)
|
|
@events = Array(events)
|
|
@remote_ip = remote_ip.to_s.presence || "unknown"
|
|
end
|
|
|
|
def call
|
|
return Result.new(accepted: 0, rejected: 0, rate_limited: true) if rate_limited?
|
|
|
|
slice = @events.first(MAX_BATCH)
|
|
accepted = 0
|
|
rejected = 0
|
|
rows = []
|
|
|
|
slice.each do |raw|
|
|
attrs = build_attrs(raw)
|
|
if attrs
|
|
rows << attrs
|
|
accepted += 1
|
|
else
|
|
rejected += 1
|
|
end
|
|
end
|
|
|
|
AnalyticsEvent.insert_all(rows) if rows.any?
|
|
if rows.any?
|
|
begin
|
|
Analytics::Aggregate.new.call
|
|
rescue ActiveRecord::RecordNotUnique => e
|
|
# Dopo i retry interni: non far fallire la richiesta analytics.
|
|
Rails.logger.warn("[analytics] aggregate race after retries, enqueue job: #{e.message}")
|
|
Analytics::AggregateJob.perform_later
|
|
end
|
|
end
|
|
|
|
Result.new(accepted: accepted, rejected: rejected + (@events.size - slice.size), rate_limited: false)
|
|
end
|
|
|
|
private
|
|
|
|
def rate_limited?
|
|
key = "analytics:ingest:#{@remote_ip}"
|
|
redis = Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0"))
|
|
count = redis.incr(key)
|
|
redis.expire(key, 60) if count == 1
|
|
count > RATE_LIMIT_PER_MINUTE
|
|
rescue StandardError => e
|
|
Rails.logger.warn("[analytics] rate limit unavailable: #{e.class}")
|
|
false
|
|
end
|
|
|
|
def build_attrs(raw)
|
|
data = raw.is_a?(Hash) ? raw.with_indifferent_access : {}
|
|
event_type = data[:type].to_s.presence || data[:event_type].to_s
|
|
return nil unless AnalyticsEvent::EVENT_TYPES.include?(event_type)
|
|
|
|
device = data[:device].to_s
|
|
return nil unless AnalyticsEvent::DEVICES.include?(device)
|
|
|
|
page_path = PathNormalizer.normalize(data[:path] || data[:page_path])
|
|
return nil if PathNormalizer.excluded?(page_path)
|
|
return nil if page_path.length > 200
|
|
|
|
occurred_at = parse_time(data[:ts] || data[:occurred_at]) || Time.current
|
|
weight = data[:n].presence || data[:weight].presence || 1
|
|
weight = Integer(weight)
|
|
weight = 1 if weight < 1
|
|
weight = 500 if weight > 500
|
|
|
|
attrs = {
|
|
page_path: page_path,
|
|
device: device,
|
|
event_type: event_type,
|
|
occurred_at: occurred_at,
|
|
weight: weight,
|
|
x_pct: nil,
|
|
y_pct: nil,
|
|
scroll_pct: nil,
|
|
created_at: Time.current,
|
|
updated_at: Time.current
|
|
}
|
|
|
|
case event_type
|
|
when "click", "move"
|
|
x = clamp_pct(data[:x] || data[:x_pct])
|
|
y = clamp_pct(data[:y] || data[:y_pct])
|
|
return nil if x.nil? || y.nil?
|
|
|
|
attrs[:x_pct] = x
|
|
attrs[:y_pct] = y
|
|
when "scroll"
|
|
scroll = clamp_pct(data[:scroll] || data[:scroll_pct])
|
|
return nil if scroll.nil?
|
|
|
|
attrs[:scroll_pct] = scroll
|
|
when "pageview"
|
|
# no extra fields
|
|
end
|
|
|
|
attrs
|
|
rescue ArgumentError, TypeError
|
|
nil
|
|
end
|
|
|
|
def clamp_pct(value)
|
|
return nil if value.nil?
|
|
|
|
n = Float(value)
|
|
return nil if n.nan? || n.infinite?
|
|
return nil if n.negative? || n > 100
|
|
|
|
n.round(2)
|
|
rescue ArgumentError, TypeError
|
|
nil
|
|
end
|
|
|
|
def parse_time(value)
|
|
return value if value.is_a?(Time) || value.is_a?(ActiveSupport::TimeWithZone)
|
|
return Time.zone.at(value.to_i / 1000.0) if value.is_a?(Numeric) || value.to_s.match?(/\A\d{10,13}\z/)
|
|
|
|
Time.zone.parse(value.to_s)
|
|
rescue ArgumentError, TypeError
|
|
nil
|
|
end
|
|
end
|
|
end
|