Reorganize event models and add comprehensive test coverage

Move InboxEvent and OutboxEvent to models directory for better
code organization. Update all require references to reflect new paths.

Add extensive test suites for:
- Consumer components (inbox processor, repository, subscription manager)
- Publisher outbox repository
- Topology management (stream, overlap guard)
- Core utilities (debug helper, model codec setup)

Enhance publisher with improved event handling and validation.
Update RuboCop configuration and spec helper.

All tests passing with 83.83% code coverage.

Author: attaradev

.rubocop.yml

@@ -7,23 +7,23 @@ plugins:
 
 AllCops:
   NewCops: enable
-  TargetRubyVersion: 3.2   # match gemspec minimum to avoid false positives
+  TargetRubyVersion: 3.2 # match gemspec minimum to avoid false positives
   SuggestExtensions: false
   Exclude:
-    - 'bin/*'
-    - 'pkg/**/*'
-    - 'tmp/**/*'
-    - 'vendor/**/*'
-    - 'spec/fixtures/**/*'
+    - "bin/*"
+    - "pkg/**/*"
+    - "tmp/**/*"
+    - "vendor/**/*"
+    - "spec/fixtures/**/*"
 
 # ---------- Layout & Style ----------
 Layout/LineLength:
   Max: 120
   Exclude:
-    - 'spec/**/*'   # allow longer expectation/setup lines
+    - "spec/**/*" # allow longer expectation/setup lines
 
 Style/Documentation:
-  Enabled: false    # don’t force top-level docs for every class/module in a gem
+  Enabled: false # don’t force top-level docs for every class/module in a gem
 
 Style/FrozenStringLiteralComment:
   Enabled: true
@@ -41,28 +41,28 @@ Style/HashSyntax:
 # ---------- Metrics (balanced for libs) ----------
 Metrics/BlockLength:
   Exclude:
-    - 'spec/**/*'
-    - 'Rakefile'
-    - 'tasks/**/*.rake'
-    - 'lib/jetstream_bridge/tasks/**/*.rake'
-    - 'jetstream_bridge.gemspec'
+    - "spec/**/*"
+    - "Rakefile"
+    - "tasks/**/*.rake"
+    - "lib/jetstream_bridge/tasks/**/*.rake"
+    - "jetstream_bridge.gemspec"
 
 Metrics/MethodLength:
   Max: 25
   Exclude:
-    - 'lib/jetstream_bridge/**/publisher*.rb'
-    - 'lib/jetstream_bridge/consumer/**/*.rb'
-    - 'lib/jetstream_bridge/core/debug_helper.rb'
-    - 'lib/jetstream_bridge/topology/stream.rb'
-    - 'lib/generators/jetstream_bridge/health_check/health_check_generator.rb'
+    - "lib/jetstream_bridge/**/publisher*.rb"
+    - "lib/jetstream_bridge/consumer/**/*.rb"
+    - "lib/jetstream_bridge/core/debug_helper.rb"
+    - "lib/jetstream_bridge/topology/stream.rb"
+    - "lib/generators/jetstream_bridge/health_check/health_check_generator.rb"
 
 Metrics/AbcSize:
   Max: 25
   Exclude:
-    - 'lib/jetstream_bridge/**/publisher*.rb'
-    - 'lib/jetstream_bridge/consumer/**/*.rb'
-    - 'lib/jetstream_bridge/core/config.rb'
-    - 'lib/jetstream_bridge/core/debug_helper.rb'
+    - "lib/jetstream_bridge/**/publisher*.rb"
+    - "lib/jetstream_bridge/consumer/**/*.rb"
+    - "lib/jetstream_bridge/core/config.rb"
+    - "lib/jetstream_bridge/core/debug_helper.rb"
 
 Metrics/CyclomaticComplexity:
   Max: 10
@@ -71,11 +71,11 @@ Metrics/PerceivedComplexity:
   Max: 10
 
 Metrics/ClassLength:
-  Max: 150
+  Max: 155
 
 Metrics/ParameterLists:
   Max: 5
-  CountKeywordArgs: false  # don't count keyword args for initialization methods
+  CountKeywordArgs: false # allow more keyword args for clarity
 
 # ---------- Naming ----------
 Naming/PredicateMethod:
@@ -86,23 +86,23 @@ Naming/PredicateMethod:
 
 # ---------- Lint ----------
 Lint/IneffectiveAccessModifier:
-  Enabled: false  # Allow class methods after private
+  Enabled: false # Allow class methods after private
 
 Lint/UselessConstantScoping:
   Enabled: false
 
 # ---------- Style ----------
 Style/MultilineBlockChain:
   Exclude:
-    - 'spec/**/*'  # allow chaining in tests for better readability
+    - "spec/**/*" # allow chaining in tests for better readability
 
 # Optional: relax just for low-level I/O integration code
 # (you can delete this once you refactor)
 
 # ---------- Packaging / Gemspec ----------
 # These ship with RuboCop core and rubocop-packaging
 Gemspec/RequiredRubyVersion:
-  Enabled: true   # aligns with your gemspec `required_ruby_version`
+  Enabled: true # aligns with your gemspec `required_ruby_version`
 
 Gemspec/DevelopmentDependencies:
   Enabled: true

lib/jetstream_bridge.rb

@@ -12,8 +12,8 @@
 require_relative 'jetstream_bridge/railtie' if defined?(Rails::Railtie)
 
 # Load gem-provided models from lib/
-require_relative 'jetstream_bridge/inbox_event'
-require_relative 'jetstream_bridge/outbox_event'
+require_relative 'jetstream_bridge/models/inbox_event'
+require_relative 'jetstream_bridge/models/outbox_event'
 
 # JetstreamBridge main module.
 module JetstreamBridge
@@ -101,6 +101,74 @@ def stream_info
       fetch_stream_info
     end
 
+    # Convenience method to publish events
+    #
+    # Supports three usage patterns:
+    #
+    # 1. Structured parameters (recommended):
+    #    JetstreamBridge.publish(resource_type: 'user', event_type: 'created', payload: { id: 1, name: 'Ada' })
+    #
+    # 2. Simplified hash (infers resource_type from event_type):
+    #    JetstreamBridge.publish(event_type: 'user.created', payload: { id: 1, name: 'Ada' })
+    #
+    # 3. Complete envelope (advanced):
+    #    JetstreamBridge.publish({ event_type: 'created', resource_type: 'user', payload: {...}, event_id: '...' })
+    #
+    # @param event_or_hash [Hash, nil] Event hash or first positional argument
+    # @param resource_type [String, nil] Resource type (e.g., 'user', 'order')
+    # @param event_type [String, nil] Event type (e.g., 'created', 'updated', 'user.created')
+    # @param payload [Hash, nil] Event payload data
+    # @param subject [String, nil] Optional subject override
+    # @param options [Hash] Additional options (event_id, occurred_at, trace_id)
+    # @return [Boolean] true if published successfully
+    def publish(event_or_hash = nil, resource_type: nil, event_type: nil, payload: nil, subject: nil, **)
+      publisher = Publisher.new
+      publisher.publish(event_or_hash, resource_type: resource_type, event_type: event_type, payload: payload,
+                                       subject: subject, **)
+    end
+
+    # Convenience method to start consuming messages
+    #
+    # Supports two usage patterns:
+    #
+    # 1. With a block (recommended):
+    #    consumer = JetstreamBridge.subscribe do |event, context|
+    #      puts "Received: #{event['event_type']}"
+    #    end
+    #    consumer.run!
+    #
+    # 2. With auto-run (returns Thread):
+    #    thread = JetstreamBridge.subscribe(run: true) do |event, context|
+    #      puts "Received: #{event['event_type']}"
+    #    end
+    #    thread.join # Wait for consumer to finish
+    #
+    # 3. With a handler object:
+    #    handler = ->(event, context) { puts event['event_type'] }
+    #    consumer = JetstreamBridge.subscribe(handler)
+    #    consumer.run!
+    #
+    # @param handler [Proc, #call, nil] Message handler (optional if block given)
+    # @param run [Boolean] If true, automatically runs consumer in a background thread
+    # @param durable_name [String, nil] Optional durable consumer name override
+    # @param batch_size [Integer, nil] Optional batch size override
+    # @yield [event, context] Yields event and context to block if no handler provided
+    # @return [Consumer, Thread] Consumer instance or Thread if run: true
+    def subscribe(handler = nil, run: false, durable_name: nil, batch_size: nil, &block)
+      handler ||= block
+      raise ArgumentError, 'Handler or block required' unless handler
+
+      consumer = Consumer.new(handler, durable_name: durable_name, batch_size: batch_size)
+
+      if run
+        thread = Thread.new { consumer.run! }
+        thread.abort_on_exception = true
+        thread
+      else
+        consumer
+      end
+    end
+
     private
 
     def fetch_stream_info

lib/jetstream_bridge/consumer/consumer.rb

@@ -19,10 +19,12 @@ class Consumer
     IDLE_SLEEP_SECS       = 0.05
     MAX_IDLE_BACKOFF_SECS = 1.0
 
-    def initialize(durable_name: nil, batch_size: nil, &block)
-      raise ArgumentError, 'handler block required' unless block_given?
+    attr_reader :durable, :batch_size
+
+    def initialize(handler = nil, durable_name: nil, batch_size: nil, &block)
+      @handler = handler || block
+      raise ArgumentError, 'handler or block required' unless @handler
 
-      @handler       = block
       @batch_size    = Integer(batch_size || DEFAULT_BATCH_SIZE)
       @durable       = durable_name || JetstreamBridge.config.durable_name
       @idle_backoff  = IDLE_SLEEP_SECS

lib/jetstream_bridge/core/connection.rb

@@ -81,7 +81,10 @@ def connect!
     # Public API for checking connection status
     # @return [Boolean] true if NATS client is connected and JetStream is healthy
     def connected?
-      @nc&.connected? && @jts && jetstream_healthy?
+      return false unless @nc&.connected?
+      return false unless @jts
+
+      jetstream_healthy?
     end
 
     # Public API for getting connection timestamp

lib/jetstream_bridge/inbox_event.rb renamed to lib/jetstream_bridge/models/inbox_event.rb

@@ -17,23 +17,92 @@ def initialize(retry_strategy: nil)
       @retry_strategy = retry_strategy || PublisherRetryStrategy.new
     end
 
+    # Publishes an event to NATS JetStream.
+    #
+    # Supports two usage patterns:
+    #
+    # 1. Structured parameters (recommended):
+    #    publish(resource_type: 'user', event_type: 'created', payload: { id: 1, name: 'Ada' })
+    #
+    # 2. Hash/envelope (advanced):
+    #    publish({ event_type: 'user.created', payload: {...}, event_id: '...', ... }, subject: 'custom.subject')
+    #
+    # @param event_or_hash [Hash] Either structured params or a complete event envelope
+    # @param resource_type [String, nil] Resource type (e.g., 'user', 'order')
+    # @param event_type [String, nil] Event type (e.g., 'created', 'updated')
+    # @param payload [Hash, nil] Event payload data
+    # @param subject [String, nil] Optional subject override
     # @return [Boolean]
-    def publish(resource_type:, event_type:, payload:, **options)
+    def publish(event_or_hash = nil, resource_type: nil, event_type: nil, payload: nil, subject: nil, **options)
       ensure_destination!
-      envelope = build_envelope(resource_type, event_type, payload, options)
-      subject  = JetstreamBridge.config.source_subject
 
+      params = { event_or_hash: event_or_hash, resource_type: resource_type, event_type: event_type,
+                 payload: payload, subject: subject, options: options }
+      envelope, resolved_subject = route_publish_params(params)
+
+      do_publish(resolved_subject, envelope)
+    rescue ArgumentError
+      # Re-raise validation errors
+      raise
+    rescue StandardError => e
+      # Only catch publishing errors
+      log_error(false, e)
+    end
+
+    def do_publish(subject, envelope)
       if JetstreamBridge.config.use_outbox
         publish_via_outbox(subject, envelope)
       else
         with_retries { publish_to_nats(subject, envelope) }
       end
-    rescue StandardError => e
-      log_error(false, e)
     end
 
     private
 
+    # Routes publish parameters to appropriate envelope builder
+    # @return [Array<Hash, String>] tuple of [envelope, subject]
+    def route_publish_params(params)
+      if structured_params?(params)
+        build_from_structured_params(params)
+      elsif keyword_or_hash_params?(params)
+        build_from_keyword_or_hash(params)
+      else
+        raise ArgumentError, 'Either provide (resource_type:, event_type:, payload:) or an event hash'
+      end
+    end
+
+    def structured_params?(params)
+      params[:resource_type] && params[:event_type] && params[:payload]
+    end
+
+    def keyword_or_hash_params?(params)
+      params[:event_type] || params[:payload] || params[:event_or_hash].is_a?(Hash)
+    end
+
+    def build_from_structured_params(params)
+      envelope = build_envelope(params[:resource_type], params[:event_type], params[:payload], params[:options])
+      resolved_subject = params[:subject] || JetstreamBridge.config.source_subject
+      [envelope, resolved_subject]
+    end
+
+    def build_from_keyword_or_hash(params)
+      envelope = if params[:event_or_hash].is_a?(Hash)
+                   normalize_envelope(params[:event_or_hash], params[:options])
+                 else
+                   build_from_keywords(params[:event_type], params[:payload], params[:options])
+                 end
+
+      resolved_subject = params[:subject] || params[:options][:subject] || JetstreamBridge.config.source_subject
+      [envelope, resolved_subject]
+    end
+
+    def build_from_keywords(event_type, payload, options)
+      raise ArgumentError, 'event_type is required' unless event_type
+      raise ArgumentError, 'payload is required' unless payload
+
+      normalize_envelope({ 'event_type' => event_type, 'payload' => payload }, options)
+    end
+
     def ensure_destination!
       return unless JetstreamBridge.config.destination_app.to_s.empty?
 
@@ -123,5 +192,36 @@ def build_envelope(resource_type, event_type, payload, options = {})
         'payload' => payload
       }
     end
+
+    # Normalize a hash to match envelope structure, allowing partial envelopes
+    def normalize_envelope(hash, options = {})
+      hash = hash.transform_keys(&:to_s)
+
+      # Allow shorthand: if only 'event_type' and 'payload', infer resource_type
+      if hash['event_type'] && hash['payload'] && !hash['resource_type']
+        # Try to infer from dot notation (e.g., 'user.created' -> 'user')
+        parts = hash['event_type'].split('.')
+        hash['resource_type'] = parts[0] if parts.size > 1
+      end
+
+      {
+        'event_id' => hash['event_id'] || options[:event_id] || SecureRandom.uuid,
+        'schema_version' => hash['schema_version'] || 1,
+        'event_type' => hash['event_type'] || raise(ArgumentError, 'event_type is required'),
+        'producer' => hash['producer'] || JetstreamBridge.config.app_name,
+        'resource_id' => hash['resource_id'] || extract_resource_id(hash['payload']),
+        'occurred_at' => hash['occurred_at'] || (options[:occurred_at] || Time.now.utc).iso8601,
+        'trace_id' => hash['trace_id'] || options[:trace_id] || SecureRandom.hex(8),
+        'resource_type' => hash['resource_type'] || 'event',
+        'payload' => hash['payload'] || raise(ArgumentError, 'payload is required')
+      }
+    end
+
+    def extract_resource_id(payload)
+      return '' unless payload
+
+      payload = payload.transform_keys(&:to_s) if payload.respond_to?(:transform_keys)
+      (payload['id'] || payload[:id] || payload['resource_id'] || payload[:resource_id]).to_s
+    end
   end
 end

lib/jetstream_bridge/outbox_event.rb renamed to lib/jetstream_bridge/models/outbox_event.rb

@@ -22,14 +22,32 @@
   after { JetstreamBridge.reset! }
 
   describe 'initialization' do
-    it 'ensures and subscribes the consumer' do
+    it 'ensures and subscribes the consumer with block' do
       described_class.new { |*| nil }
       expect(JetstreamBridge::SubscriptionManager)
         .to have_received(:new)
         .with(jts, JetstreamBridge.config.durable_name, JetstreamBridge.config)
       expect(sub_mgr).to have_received(:ensure_consumer!)
       expect(sub_mgr).to have_received(:subscribe!)
     end
+
+    it 'accepts handler as first argument' do
+      handler = ->(event, context) { nil }
+      consumer = described_class.new(handler)
+      expect(consumer).to be_a(described_class)
+    end
+
+    it 'raises error when neither handler nor block provided' do
+      expect do
+        described_class.new
+      end.to raise_error(ArgumentError, /handler or block required/)
+    end
+
+    it 'accepts custom durable_name and batch_size' do
+      consumer = described_class.new(durable_name: 'custom-durable', batch_size: 50) { |*| nil }
+      expect(consumer.durable).to eq('custom-durable')
+      expect(consumer.batch_size).to eq(50)
+    end
   end
 
   describe '#process_batch' do