Add latest changes from gitlab-org/gitlab@master

4 files changed, 181 insertions, 0 deletions

diff --git a/lib/gitlab/event_store/event.rb b/lib/gitlab/event_store/event.rb
new file mode 100644
index 00000000000..ee0c329b8e8
--- /dev/null
+++ b/lib/gitlab/event_store/event.rb
@@ -0,0 +1,54 @@
+# frozen_string_literal: true
+
+# An Event object represents a domain event that occurred in a bounded context.
+# By publishing events we notify other bounded contexts about something
+# that happened, so that they can react to it.
+#
+# Define new event classes under `app/events/<namespace>/` with a name
+# representing something that happened in the past:
+#
+#   class Projects::ProjectCreatedEvent < Gitlab::EventStore::Event
+#     def schema
+#       {
+#         'type' => 'object',
+#         'properties' => {
+#           'project_id' => { 'type' => 'integer' }
+#         }
+#       }
+#     end
+#   end
+#
+# To publish it:
+#
+#   Gitlab::EventStore.publish(
+#     Projects::ProjectCreatedEvent.new(data: { project_id: project.id })
+#   )
+#
+module Gitlab
+  module EventStore
+    class Event
+      attr_reader :data
+
+      def initialize(data:)
+        validate_schema!(data)
+        @data = data
+      end
+
+      def schema
+        raise NotImplementedError, 'must specify schema to validate the event'
+      end
+
+      private
+
+      def validate_schema!(data)
+        unless data.is_a?(Hash)
+          raise Gitlab::EventStore::InvalidEvent, "Event data must be a Hash"
+        end
+
+        unless JSONSchemer.schema(schema).valid?(data.deep_stringify_keys)
+          raise Gitlab::EventStore::InvalidEvent, "Data for event #{self.class} does not match the defined schema: #{schema}"
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/event_store/store.rb b/lib/gitlab/event_store/store.rb
new file mode 100644
index 00000000000..ecf3cd7e562
--- /dev/null
+++ b/lib/gitlab/event_store/store.rb
@@ -0,0 +1,54 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module EventStore
+    class Store
+      attr_reader :subscriptions
+
+      def initialize
+        @subscriptions = Hash.new { |h, k| h[k] = [] }
+
+        yield(self) if block_given?
+
+        # freeze the subscriptions as safety measure to avoid further
+        # subcriptions after initialization.
+        lock!
+      end
+
+      def subscribe(worker, to:, if: nil)
+        condition = binding.local_variable_get('if')
+
+        Array(to).each do |event|
+          validate_subscription!(worker, event)
+          subscriptions[event] << Gitlab::EventStore::Subscription.new(worker, condition)
+        end
+      end
+
+      def publish(event)
+        unless event.is_a?(Event)
+          raise InvalidEvent, "Event being published is not an instance of Gitlab::EventStore::Event: got #{event.inspect}"
+        end
+
+        subscriptions[event.class].each do |subscription|
+          subscription.consume_event(event)
+        end
+      end
+
+      private
+
+      def lock!
+        @subscriptions.freeze
+      end
+
+      def validate_subscription!(subscriber, event_class)
+        unless event_class < Event
+          raise InvalidEvent, "Event being subscribed to is not a subclass of Gitlab::EventStore::Event: got #{event_class}"
+        end
+
+        unless subscriber.respond_to?(:perform_async)
+          raise InvalidSubscriber, "Subscriber is not an ApplicationWorker: got #{subscriber}"
+        end
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/event_store/subscriber.rb b/lib/gitlab/event_store/subscriber.rb
new file mode 100644
index 00000000000..cf326d1f9e4
--- /dev/null
+++ b/lib/gitlab/event_store/subscriber.rb
@@ -0,0 +1,36 @@
+# frozen_string_literal: true
+
+# This module should be included in order to turn an ApplicationWorker
+# into a Subscriber.
+# This module overrides the `perform` method and provides a better and
+# safer interface for handling events via `handle_event` method.
+#
+# @example:
+#   class SomeEventSubscriber
+#     include ApplicationWorker
+#     include Gitlab::EventStore::Subscriber
+#
+#     def handle_event(event)
+#       # ...
+#     end
+#   end
+
+module Gitlab
+  module EventStore
+    module Subscriber
+      def perform(event_type, data)
+        raise InvalidEvent, event_type unless self.class.const_defined?(event_type)
+
+        event = event_type.constantize.new(
+          data: data.with_indifferent_access
+        )
+
+        handle_event(event)
+      end
+
+      def handle_event(event)
+        raise NotImplementedError, 'you must implement this methods in order to handle events'
+      end
+    end
+  end
+end
diff --git a/lib/gitlab/event_store/subscription.rb b/lib/gitlab/event_store/subscription.rb
new file mode 100644
index 00000000000..e5c92ab969f
--- /dev/null
+++ b/lib/gitlab/event_store/subscription.rb
@@ -0,0 +1,37 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module EventStore
+    class Subscription
+      attr_reader :worker, :condition
+
+      def initialize(worker, condition)
+        @worker = worker
+        @condition = condition
+      end
+
+      def consume_event(event)
+        return unless condition_met?(event)
+
+        worker.perform_async(event.class.name, event.data)
+        # TODO: Log dispatching of events to subscriber
+
+        # We rescue and track any exceptions here because we don't want to
+        # impact other subscribers if one is faulty.
+        # The method `condition_met?`, since it can run a block, it might encounter
+        # a bug. By raising an exception here we could interrupt the publishing
+        # process, preventing other subscribers from consuming the event.
+      rescue StandardError => e
+        Gitlab::ErrorTracking.track_and_raise_for_dev_exception(e, event_class: event.class.name, event_data: event.data)
+      end
+
+      private
+
+      def condition_met?(event)
+        return true unless condition
+
+        condition.call(event)
+      end
+    end
+  end
+end