Class: Webhookdb::Replicator::IcalendarCalendarV1::EventProcessor

Inherits:

Object

• Object
• Webhookdb::Replicator::IcalendarCalendarV1::EventProcessor

Defined in:

lib/webhookdb/replicator/icalendar_calendar_v1.rb

Instance Attribute Summary

• #upserted_identities ⇒ Object readonly

  Returns the value of attribute upserted_identities.

Instance Method Summary

• #_ical_entry_from_ruby(r, entry, is_date) ⇒ Object

  We need is_date because the recurrence/IceCube schedule may be using times, not date.

• #_icecube_rule_from_ical(ical) ⇒ Object
• #_time_array(h) ⇒ Object
• #delete_condition ⇒ Object
• #each_feed_event ⇒ Object
• #each_projected_event(h) ⇒ Object
• #initialize(io, upserter) ⇒ EventProcessor constructor

  A new instance of EventProcessor.

• #process ⇒ Object

Constructor Details

#initialize(io, upserter) ⇒ EventProcessor

Returns a new instance of EventProcessor.

# File 'lib/webhookdb/replicator/icalendar_calendar_v1.rb', line 246

def initialize(io, upserter)
  @io = io
  @upserter = upserter
  # Keep track of everything we upsert. For any rows we aren't upserting,
  # delete them if they're recurring, or cancel them if they're not recurring.
  # If doing it this way is slow, we could invert this (pull down all IDs and pop from the set).
  @upserted_identities = []
  # Keep track of all upserted recurring items.
  # If we find a RECURRENCE-ID on a later item,
  # we need to modify the item from the sequence by stealing its compound identity.
  @expanded_events_by_uid = {}
  # Delete 'extra' recurring event rows.
  # We need to keep track of how many events each UID spawns,
  # so we can delete any with a higher count.
  @max_sequence_num_by_uid = {}
end

Instance Attribute Details

#upserted_identities ⇒ Object (readonly)

Returns the value of attribute upserted_identities.

# File 'lib/webhookdb/replicator/icalendar_calendar_v1.rb', line 244

def upserted_identities
  @upserted_identities
end

Instance Method Details

#_ical_entry_from_ruby(r, entry, is_date) ⇒ Object

We need is_date because the recurrence/IceCube schedule may be using times, not date.

# File 'lib/webhookdb/replicator/icalendar_calendar_v1.rb', line 408

def _ical_entry_from_ruby(r, entry, is_date)
  return {"v" => r.strftime("%Y%m%d")} if is_date
  return {"v" => r.strftime("%Y%m%dT%H%M%SZ")} if r.zone == "UTC"
  return {"v" => r.strftime("%Y%m%dT%H%M%S"), "TZID" => entry.fetch("TZID")}
end

#_icecube_rule_from_ical(ical) ⇒ Object

# File 'lib/webhookdb/replicator/icalendar_calendar_v1.rb', line 414

def _icecube_rule_from_ical(ical)
  # We have seen certain ambiguous rules, like FREQ=WEEKLY with BYMONTHDAY=4.
  # Apple interprets this as every 2 weeks; rrule.js interprets it as on the 4th of the month.
  # IceCube errors, because `day_of_month` isn't valid on a WeeklyRule.
  # In this case, we need to sanitize the string to remove the offending rule piece.
  # There are probably many other offending formats, but we'll add them here as needed.
  if ical.include?("FREQ=WEEKLY") && ical.include?("BYMONTHDAY=")
    ical = ical.gsub(/BYMONTHDAY=[\d,]+/, "")
    ical.delete_prefix! ";"
    ical.delete_suffix! ";"
  end
  return IceCube::IcalParser.rule_from_ical(ical)
end

#_time_array(h) ⇒ Object

# File 'lib/webhookdb/replicator/icalendar_calendar_v1.rb', line 428

def _time_array(h)
  expanded_entries = h["v"].split(",").map { |v| h.merge("v" => v) }
  return expanded_entries.map do |e|
    parsed_val, _got_tz = Webhookdb::Replicator::IcalendarEventV1.entry_to_date_or_datetime(e)
    next parsed_val if parsed_val.is_a?(Date)
    # Convert to UTC. We don't work with ActiveSupport timezones in the icalendar code for the most part.
    parsed_val.utc
  end
end

#delete_condition ⇒ Object

# File 'lib/webhookdb/replicator/icalendar_calendar_v1.rb', line 263

def delete_condition
  return nil if @max_sequence_num_by_uid.empty?
  return @max_sequence_num_by_uid.map do |uid, n|
    Sequel[recurring_event_id: uid] & (Sequel[:recurring_event_sequence] > n)
  end.inject(&:|)
end

#each_feed_event ⇒ Object

# File 'lib/webhookdb/replicator/icalendar_calendar_v1.rb', line 438

def each_feed_event
  bad_event_uids = Set.new
  vevent_lines = []
  in_vevent = false
  while (line = @io.gets)
    line.rstrip!
    if line == "BEGIN:VEVENT"
      in_vevent = true
      vevent_lines << line
    elsif line == "END:VEVENT"
      in_vevent = false
      vevent_lines << line
      h = Webhookdb::Replicator::IcalendarEventV1.vevent_to_hash(vevent_lines)
      vevent_lines.clear
      if h.key?("DTSTART") && h.key?("UID")
        yield h
      else
        bad_event_uids << h.fetch("UID", {}).fetch("v", "[missing]")
      end
    elsif in_vevent
      vevent_lines << line
    end
  end
  return if bad_event_uids.empty?
  @upserter.upserting_replicator.logger.warn("invalid_vevent_hash", vevent_uids: bad_event_uids.sort)
end

#each_projected_event(h) ⇒ Object

Raises:

• (LocalJumpError)

# File 'lib/webhookdb/replicator/icalendar_calendar_v1.rb', line 284

def each_projected_event(h)
  raise LocalJumpError unless block_given?

  uid = h.fetch("UID").fetch("v")

  if (recurrence_id = h["RECURRENCE-ID"])
    # Track down the original item in the projected sequence, so we can update it.
    if Webhookdb::Replicator::IcalendarEventV1.value_is_date_str?(recurrence_id.fetch("v"))
      start = Webhookdb::Replicator::IcalendarEventV1.entry_to_date(recurrence_id)
      startfield = :start_date
    else
      startfield = :start_at
      start = Webhookdb::Replicator::IcalendarEventV1.entry_to_datetime(recurrence_id).first
    end
    candidates = @expanded_events_by_uid[uid]
    if candidates.nil?
      # We can have no recurring events, even with the exclusion date.
      # Not much we can do here- just treat it as a standalone event.
      yield h
      return
    end
    unless (match = candidates.find { |c| c[startfield] == start })
      # There are some providers (like Apple) where an excluded event
      # will be outside the bounds of the RRULE of its owner.
      # Usually the RRULE has an UNTIL that is before the RECURRENCE-ID datetime.
      #
      # In these cases, we can use the event as-is, but we need to
      # make sure it is treated as part of the sequence.
      # So increment the last-seen sequence number for the UID and use that.
      max_seq_num = @max_sequence_num_by_uid[uid] += 1
      h["UID"] = {"v" => "#{uid}-#{max_seq_num}"}
      h["recurring_event_id"] = uid
      h["recurring_event_sequence"] = max_seq_num
      yield h
      return
    end

    # Steal the UID to overwrite the original, and record where it came from.
    # Note that all other fields, like categories, will be overwritten with the fields in this exclusion.
    # This seems to be correct, but we should keep an eye open in case we need to merge
    # these exclusion events into the originals.
    h["UID"] = {"v" => match[:uid]}
    h["recurring_event_sequence"] = match[:recurring_event_sequence]
    # Usually the recurrent event and exclusion have the same last-modified.
    # But we need to set the last-modified to AFTER the original,
    # to make sure it replaces what's in the database (the original un-excluded event
    # may already be present in the database).
    h["LAST-MODIFIED"] = match.fetch(:last_modified_at) + 1.second
    yield h
    return
  end

  unless h["RRULE"]
    yield h
    return
  end

  # We need to convert relevant parsed ical lines back to a string for use in ice_cube.
  # There are other ways to handle this, but this is fine for now.
  ical_params = {}
  if (exdates = h["RDATE"])
    ical_params[:rtimes] = exdates.map { |d| self._time_array(d) }.flatten
  end
  if (exdates = h["EXDATE"])
    ical_params[:extimes] = exdates.map { |d| self._time_array(d) }.flatten
  end
  ical_params[:rrules] = [self._icecube_rule_from_ical(h["RRULE"]["v"])] if h["RRULE"]
  # DURATION is not supported

  start_entry = h.fetch("DTSTART")
  ev_replicator = Webhookdb::Replicator::IcalendarEventV1
  is_date = ev_replicator.entry_is_date_str?(start_entry)
  # Use actual Times for start/end since ice_cube doesn't parse them well
  ical_params[:start_time] = ev_replicator.entry_to_date_or_datetime(start_entry).first
  if ical_params[:start_time].year < 1000
    # This is almost definitely a misconfiguration. Yield it as non-recurring and move on.
    yield h
    return
  end
  has_end_time = false
  if (end_entry = h["DTEND"])
    # the end date is optional. If we don't have one, we should never store one.
    has_end_time = true
    ical_params[:end_time] = ev_replicator.entry_to_date_or_datetime(end_entry).first
    if ical_params[:end_time] < ical_params[:start_time]
      # This is an invalid event. Not sure what it'll do to IceCube so don't send it there.
      # Yield it as a non-recurring event and move on.
      yield h
      return
    end
  end

  schedule = IceCube::Schedule.from_hash(ical_params)
  dont_project_before = Webhookdb::Icalendar.oldest_recurring_event
  dont_project_after = Time.now + RECURRENCE_PROJECTION

  # Just like google, track the original event id.
  h["recurring_event_id"] = uid
  final_sequence = -1
  begin
    schedule.send(:enumerate_occurrences, schedule.start_time).each_with_index do |occ, idx|
      next if occ.start_time < dont_project_before
      # Given the original hash, we will modify some fields.
      e = h.dup
      # Keep track of how many events we're managing.
      e["recurring_event_sequence"] = idx
      # The new UID has the sequence number.
      e["UID"] = {"v" => "#{uid}-#{idx}"}
      e["DTSTART"] = self._ical_entry_from_ruby(occ.start_time, start_entry, is_date)
      e["DTEND"] = self._ical_entry_from_ruby(occ.end_time, end_entry, is_date) if has_end_time
      yield e
      final_sequence = idx
      break if occ.start_time > dont_project_after
    end
  rescue Date::Error
    # It's possible we yielded some recurring events too, in that case, treat them as normal,
    # in addition to yielding the event as non-recurring.
    yield h
  end
  @max_sequence_num_by_uid[uid] = final_sequence
  return
end

#process ⇒ Object

# File 'lib/webhookdb/replicator/icalendar_calendar_v1.rb', line 270

def process
  self.each_feed_event do |feed_event|
    self.each_projected_event(feed_event) do |ev|
      ident, upserted = @upserter.handle_item(ev)
      @upserted_identities << ident
      if (recurring_uid = upserted.fetch(:recurring_event_id))
        @expanded_events_by_uid[recurring_uid] ||= []
        @expanded_events_by_uid[recurring_uid] << upserted
      end
    end
  end
  @upserter.flush_pending_inserts
end