feat: implement webhook signing and delivery (#482)

* feat: implement webhook signing and delivery

- Introduced Karya::OutboundEvents module to handle versioned outbound events.
- Added WebhookSigner and WebhookVerifier classes for signing and verifying webhook payloads.
- Implemented CloudEvents-compatible JSON envelope for outbound events.
- Created Dispatcher class to manage delivery of signed events.
- Enhanced Worker and WorkerSupervisor runtimes to support outbound event dispatching.
- Added tests for outbound event functionality, including signing, verification, and error handling.
- Updated documentation to include outbound events and webhook signing conventions.

closes: #372

Author: niteshpurohit

core/karya/lib/karya.rb

@@ -17,6 +17,7 @@
 require_relative 'karya/job'
 require_relative 'karya/retry_policy'
 require_relative 'karya/retry_policy_set'
+require_relative 'karya/outbound_events'
 require_relative 'karya/reservation'
 require_relative 'karya/queue_store'
 require_relative 'karya/workflow'

core/karya/lib/karya/base.rb

@@ -19,14 +19,28 @@ class Error < StandardError; end
   # Raised when runtime code requires a configured queue store but none has been set.
   class MissingQueueStoreConfigurationError < Error; end
 
+  # Raised when outbound event input cannot be normalized into a supported contract.
+  class InvalidOutboundEventError < Error; end
+
+  # Raised when a caller asks for an outbound event that is not part of the supported contract.
+  class UnsupportedOutboundEventError < Error; end
+
+  # Raised when a webhook signature cannot be parsed or verified.
+  class InvalidWebhookSignatureError < Error; end
+
   class << self
-    attr_reader :instrumenter
+    attr_reader :instrumenter, :outbound_event_dispatcher
 
     def configure_instrumenter(instrumenter)
       # Process-wide default used when a runtime does not receive an explicit instrumenter.
       @instrumenter = instrumenter
     end
 
+    def configure_outbound_event_dispatcher(outbound_event_dispatcher)
+      # Process-wide default used when a runtime does not receive an explicit outbound event dispatcher.
+      @outbound_event_dispatcher = outbound_event_dispatcher
+    end
+
     def configure_logger(logger)
       # Process-wide default used when a runtime does not receive an explicit logger.
       @logger = logger

core/karya/lib/karya/internal/hook_dispatch.rb

@@ -0,0 +1,51 @@
+# frozen_string_literal: true
+
+# Copyright Codevedas Inc. 2025-present
+#
+# This source code is licensed under the MIT license found in the
+# LICENSE file in the root directory of this source tree.
+
+module Karya
+  module Internal
+    # Shares runtime hook payload normalization and dispatch flow.
+    class HookDispatch
+      def self.instrument(
+        event:,
+        payload:,
+        payload_keywords:,
+        payload_given:,
+        instrumenter:,
+        dispatch_outbound:,
+        error_class:,
+        mixed_payload_message:,
+        emit_instrumentation:,
+        emit_outbound_event:
+      )
+        return nil unless instrumenter || dispatch_outbound
+
+        normalized_payload = PayloadInput.new(
+          payload,
+          payload_keywords,
+          payload_given:,
+          error_class:,
+          mixed_payload_message:
+        ).to_h
+
+        if instrumenter && dispatch_outbound
+          instrumentation_payload, outbound_payload = ImmutableHookPayload.snapshot_pair(
+            normalized_payload,
+            error_class:
+          )
+          emit_instrumentation.call(event, instrumentation_payload)
+          emit_outbound_event.call(event, outbound_payload)
+          return nil
+        end
+
+        snapshot = ImmutableHookPayload.snapshot(normalized_payload, error_class:)
+        emit_instrumentation.call(event, snapshot) if instrumenter
+        emit_outbound_event.call(event, snapshot) if dispatch_outbound
+        nil
+      end
+    end
+  end
+end

core/karya/lib/karya/internal/immutable_hook_payload.rb

@@ -0,0 +1,77 @@
+# frozen_string_literal: true
+
+# Copyright Codevedas Inc. 2025-present
+#
+# This source code is licensed under the MIT license found in the
+# LICENSE file in the root directory of this source tree.
+
+module Karya
+  module Internal
+    # Builds immutable snapshots for runtime hook payloads.
+    class ImmutableHookPayload
+      def self.snapshot(payload, error_class:)
+        new(payload, error_class:).snapshot
+      end
+
+      def self.snapshot_pair(payload, error_class:)
+        snapshot = snapshot(payload, error_class:)
+        [snapshot, shallow_snapshot(snapshot)].freeze
+      end
+
+      def self.snapshot_key(value)
+        return value if value.is_a?(Symbol)
+        return value.frozen? ? value : value.dup.freeze if value.is_a?(String)
+
+        raise ArgumentError, 'payload keys must be Symbols or Strings'
+      end
+      private_class_method :snapshot_key
+
+      def initialize(payload, error_class:)
+        @payload = payload
+        @error_class = error_class
+      end
+
+      def snapshot
+        snapshot_hash(payload)
+      end
+
+      private
+
+      attr_reader :error_class, :payload
+
+      def self.shallow_snapshot(snapshot)
+        snapshot.each_with_object({}) do |(key, value), duplicated|
+          duplicated[key] = value
+        end.freeze
+      end
+      private_class_method :shallow_snapshot
+
+      def snapshot_hash(value)
+        value.each_with_object({}) do |(key, item), duplicated|
+          duplicated[self.class.send(:snapshot_key, key)] = snapshot_value(item)
+        rescue ArgumentError => e
+          raise error_class, e.message
+        end.freeze
+      end
+
+      def snapshot_array(value)
+        value.map { |item| snapshot_value(item) }.freeze
+      end
+
+      def snapshot_value(value)
+        case value
+        when Hash
+          snapshot_hash(value)
+        when Array
+          snapshot_array(value)
+        when String, Time
+          value.frozen? ? value : value.dup.freeze
+        when NilClass, TrueClass, FalseClass, Numeric, Symbol
+          value
+        else
+          raise error_class, 'payload values must be nil, booleans, numerics, strings, symbols, times, arrays, or hashes'
+        end
+      end
+    end
+  end
+end

core/karya/lib/karya/internal/payload_input.rb

@@ -0,0 +1,53 @@
+# frozen_string_literal: true
+
+# Copyright Codevedas Inc. 2025-present
+#
+# This source code is licensed under the MIT license found in the
+# LICENSE file in the root directory of this source tree.
+
+module Karya
+  module Internal
+    # Normalizes positional and keyword payload inputs into one Hash.
+    class PayloadInput
+      # Unique sentinel for omitted positional payload arguments.
+      class Absent
+        def self.instance
+          @instance ||= new.freeze
+        end
+
+        private_class_method :new
+      end
+      private_constant :Absent
+
+      ABSENT = Absent.instance
+
+      def initialize(payload, payload_keywords, payload_given:, error_class:, mixed_payload_message:)
+        @payload = payload
+        @payload_keywords = payload_keywords
+        @payload_given = payload_given
+        @error_class = error_class
+        @mixed_payload_message = mixed_payload_message
+      end
+
+      def to_h
+        return payload_keywords unless payload_given
+
+        payload_is_hash = payload.is_a?(Hash)
+
+        if payload_keywords.empty?
+          raise error_class, 'payload must be a Hash' unless payload_is_hash
+
+          return payload
+        end
+
+        raise error_class, mixed_payload_message unless payload_is_hash
+
+        payload.merge(payload_keywords)
+      end
+
+      private
+
+      attr_reader :error_class, :mixed_payload_message, :payload, :payload_given, :payload_keywords
+    end
+  end
+end

core/karya/lib/karya/outbound_events.rb

@@ -0,0 +1,23 @@
+# frozen_string_literal: true
+
+# Copyright Codevedas Inc. 2025-present
+#
+# This source code is licensed under the MIT license found in the
+# LICENSE file in the root directory of this source tree.
+
+require_relative 'base'
+require_relative 'outbound_events/values'
+require_relative 'outbound_events/delivery'
+require_relative 'outbound_events/dispatcher'
+require_relative 'outbound_events/event'
+require_relative 'outbound_events/schema'
+require_relative 'outbound_events/schema_catalog'
+require_relative 'outbound_events/webhook_signature'
+require_relative 'outbound_events/webhook_signer'
+require_relative 'outbound_events/webhook_verifier'
+
+module Karya
+  # Shared outbound event contracts for external delivery and verification.
+  module OutboundEvents
+  end
+end

core/karya/lib/karya/outbound_events/delivery.rb

@@ -0,0 +1,75 @@
+# frozen_string_literal: true
+
+require_relative 'event'
+require_relative 'webhook_signature'
+
+# Copyright Codevedas Inc. 2025-present
+#
+# This source code is licensed under the MIT license found in the
+# LICENSE file in the root directory of this source tree.
+
+module Karya
+  module OutboundEvents
+    # Immutable serialized outbound delivery with canonical headers and body.
+    class Delivery
+      CONTENT_TYPE = 'application/cloudevents+json'
+
+      attr_reader :body, :event, :headers, :signature
+
+      def initialize(event:, signature: nil, body: nil)
+        @event = normalize_event(event)
+        @body = normalize_body(body)
+        @signature = normalize_signature(signature)
+        @headers = build_headers.freeze
+        freeze
+      end
+
+      private
+
+      def normalize_event(value)
+        return value if value.is_a?(Event)
+
+        raise InvalidOutboundEventError, 'event must be Karya::OutboundEvents::Event'
+      end
+
+      def normalize_signature(value)
+        return nil if [nil].include?(value)
+        return value if value.is_a?(WebhookSignature)
+
+        raise InvalidOutboundEventError, 'signature must be Karya::OutboundEvents::WebhookSignature'
+      end
+
+      def normalize_body(value)
+        Body.new(value, event: @event).normalize
+      end
+
+      def build_headers
+        { 'Content-Type' => CONTENT_TYPE }.merge(signature&.headers || {})
+      end
+
+      # Normalizes one optional serialized body value for an outbound delivery.
+      class Body
+        def initialize(value, event:)
+          @value = value
+          @event = event
+        end
+
+        def normalize
+          return event.to_json.freeze if [nil].include?(value)
+
+          string_value = value if value.is_a?(String)
+          return string_value if string_value&.frozen?
+          return string_value.dup.freeze if string_value
+
+          raise InvalidOutboundEventError, 'body must be a String'
+        end
+
+        private
+
+        attr_reader :event, :value
+      end
+
+      private_constant :Body
+    end
+  end
+end

core/karya/lib/karya/outbound_events/dispatcher.rb

@@ -0,0 +1,69 @@
+# frozen_string_literal: true
+
+require 'securerandom'
+require_relative '../internal/payload_input'
+require_relative '../primitives/callable'
+require_relative 'delivery'
+require_relative 'schema_catalog'
+require_relative 'webhook_signer'
+
+# Copyright Codevedas Inc. 2025-present
+#
+# This source code is licensed under the MIT license found in the
+# LICENSE file in the root directory of this source tree.
+
+module Karya
+  module OutboundEvents
+    # Builds canonical outbound deliveries from runtime instrumentation events.
+    class Dispatcher
+      def initialize(delivery_handler:, signer: nil, clock: -> { Time.now.utc }, event_id_generator: -> { SecureRandom.uuid })
+        @delivery_handler = Primitives::Callable.new(:delivery_handler, delivery_handler, error_class: InvalidOutboundEventError).normalize
+        @signer = normalize_signer(signer)
+        @clock = Primitives::Callable.new(:clock, clock, error_class: InvalidOutboundEventError).normalize
+        @event_id_generator = Primitives::Callable.new(
+          :event_id_generator,
+          event_id_generator,
+          error_class: InvalidOutboundEventError
+        ).normalize
+      end
+
+      def call(event_name, payload = Internal::PayloadInput::ABSENT, **payload_keywords)
+        return nil unless SchemaCatalog.supported?(event_name)
+
+        occurred_at = clock.call
+        raise InvalidOutboundEventError, 'clock must return a Time' unless occurred_at.is_a?(Time)
+
+        payload_given = !payload.equal?(Internal::PayloadInput::ABSENT)
+
+        event = SchemaCatalog.build_event(
+          event_name:,
+          payload: Internal::PayloadInput.new(
+            payload_given ? payload : nil,
+            payload_keywords,
+            payload_given:,
+            error_class: InvalidOutboundEventError,
+            mixed_payload_message: 'payload must be a Hash when keyword payload is also given'
+          ).to_h,
+          occurred_at:,
+          event_id: event_id_generator.call
+        )
+        body = event.to_json.freeze
+        signature = signer&.sign(body:, now: occurred_at)
+        delivery = Delivery.new(event:, signature:, body:)
+        delivery_handler.call(delivery)
+        delivery
+      end
+
+      private
+
+      attr_reader :clock, :delivery_handler, :event_id_generator, :signer
+
+      def normalize_signer(value)
+        return nil if [nil].include?(value)
+        return value if value.is_a?(WebhookSigner)
+
+        raise InvalidOutboundEventError, 'signer must be Karya::OutboundEvents::WebhookSigner'
+      end
+    end
+  end
+end