Module: Acta

Defined in:
lib/acta.rb,
lib/acta/web.rb,
lib/acta/actor.rb,
lib/acta/event.rb,
lib/acta/model.rb,
lib/acta/errors.rb,
lib/acta/record.rb,
lib/acta/schema.rb,
lib/acta/command.rb,
lib/acta/current.rb,
lib/acta/handler.rb,
lib/acta/railtie.rb,
lib/acta/reactor.rb,
lib/acta/testing.rb,
lib/acta/version.rb,
lib/acta/adapters.rb,
lib/acta/upcaster.rb,
lib/acta/projection.rb,
lib/acta/web/engine.rb,
lib/acta/reactor_job.rb,
lib/acta/testing/dsl.rb,
lib/acta/types/array.rb,
lib/acta/types/model.rb,
lib/acta/events_query.rb,
lib/acta/serializable.rb,
lib/acta/adapters/base.rb,
lib/acta/adapters/sqlite.rb,
lib/acta/web/events_query.rb,
lib/acta/adapters/postgres.rb,
lib/acta/projection_managed.rb,
lib/acta/types/encrypted_string.rb,
app/helpers/acta/web/application_helper.rb,
app/controllers/acta/web/events_controller.rb,
lib/generators/acta/install/install_generator.rb,
app/controllers/acta/web/application_controller.rb

Defined Under Namespace

Modules: Adapters, Generators, ProjectionManaged, Schema, Serializable, Testing, Types, Upcaster, Web Classes: Actor, AdapterError, Command, CommandError, ConfigurationError, Current, Error, Event, EventsQuery, EventsRecord, FutureSchemaVersion, Handler, InvalidCommand, InvalidEvent, MissingActor, Model, Projection, ProjectionError, ProjectionWriteError, Railtie, Reactor, ReactorJob, Record, ReplayError, ReplayHaltedByUpcaster, TruncateOrderError, UnknownEventType, UpcasterRegistryError, VersionConflict

Constant Summary collapse

VERSION = 
"0.4.0.alpha.1"

Class Attribute Summary collapse

Class Method Summary collapse

Class Attribute Details

.reactor_queue ⇒ Object 



38
39
40
 
# File 'lib/acta.rb', line 38

def self.reactor_queue
  @reactor_queue
end

Class Method Details

.adapter ⇒ Object 



42
43
44
 
# File 'lib/acta.rb', line 42

def self.adapter
  @adapter ||= Adapters.for(Record.connection)
end

.dispatch(event, kind: nil) ⇒ Object 



77
78
79
80
81
82
83
84
85
86
87
 
# File 'lib/acta.rb', line 77

def self.dispatch(event, kind: nil)
  handlers.each do |event_class, registrations|
    next unless event.is_a?(event_class)

    registrations.each do |registration|
      next if kind && registration[:kind] != kind

      invoke(event, registration)
    end
  end
end

.emit(event, actor: nil, if_version: nil) ⇒ Object

Raises:



50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/acta.rb', line 50

def self.emit(event, actor: nil, if_version: nil)
  event.actor = actor if actor
  raise MissingActor, "No actor for emit of #{event.event_type} (set Acta::Current.actor or pass actor:)" if event.actor.nil?

  assert_version!(event, if_version) unless if_version.nil?

  ActiveSupport::Notifications.instrument("acta.event_emitted", event:, event_type: event.event_type) do
    Record.transaction(requires_new: true) do
      record = adapter.insert_event(record_attributes_for(event))
      event.recorded_at = record.recorded_at
      dispatch(event, kind: :projection)
    end
    dispatch(event, kind: :handler)
    dispatch(event, kind: :reactor)
  end

  event
end

.events ⇒ Object 



302
303
304
 
# File 'lib/acta.rb', line 302

def self.events
  EventsQuery.new(adapter.fetch_records)
end

.handlers ⇒ Object 



73
74
75
 
# File 'lib/acta.rb', line 73

def self.handlers
  @handlers ||= Hash.new { |h, k| h[k] = [] }
end

.projection_classes ⇒ Object 



151
152
153
 
# File 'lib/acta.rb', line 151

def self.projection_classes
  @projection_classes ||= []
end

.rebuild! ⇒ Object 



174
175
176
177
178
179
180
181
182
183
184
185
186
 
# File 'lib/acta.rb', line 174

def self.rebuild!
  Projection.applying! { truncate_projections! }
  context = Upcaster::Context.new
  Record.order(:id).find_each do |record|
    events.upcast_and_hydrate(record, context).each do |event|
      dispatch(event, kind: :projection)
    end
  rescue ProjectionError, ReplayHaltedByUpcaster, FutureSchemaVersion
    raise
  rescue StandardError => e
    raise ReplayError.new(record:, original: e)
  end
end

.register_projection(klass) ⇒ Object 



155
156
157
 
# File 'lib/acta.rb', line 155

def self.register_projection(klass)
  projection_classes << klass unless projection_classes.include?(klass)
end

.register_upcaster(klass) ⇒ Object

Register a set of upcasters (a module/class that ‘include Acta::Upcaster` and declares `upcasts(…)` blocks). Idempotent — re-registering the same class is a no-op. See `Acta::Upcaster`.



162
163
164
 
# File 'lib/acta.rb', line 162

def self.register_upcaster(klass)
  upcaster_registry.register(klass)
end

.reset_adapter! ⇒ Object 



46
47
48
 
# File 'lib/acta.rb', line 46

def self.reset_adapter!
  @adapter = nil
end

.reset_handlers! ⇒ Object 



146
147
148
149
 
# File 'lib/acta.rb', line 146

def self.reset_handlers!
  @handlers = Hash.new { |h, k| h[k] = [] }
  @projection_classes = []
end

.reset_upcasters! ⇒ Object 



170
171
172
 
# File 'lib/acta.rb', line 170

def self.reset_upcasters!
  upcaster_registry.clear!
end

.set_events_record_parent!(parent) ⇒ Object

Re-parent EventsRecord (and therefore Record) onto a host-supplied abstract class. Must run BEFORE any query against Acta::Record executes — call from a host initializer after the parent class is defined. Re-defines the two constants so existing references to ‘Acta::Record` resolve to the new class.

Use case: per-tenant SQLite sharding where the host wants events and its own tenant-scoped rows in the same connection pool to avoid SQLite write contention on cross-pool transactions.

Raises:

  • (ArgumentError) 


46
47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/acta/record.rb', line 46

def self.set_events_record_parent!(parent)
  raise ArgumentError, "parent must be an abstract ActiveRecord class" unless parent.is_a?(Class) && parent < ::ActiveRecord::Base && parent.abstract_class?

  Acta.send(:remove_const, :Record)        if Acta.const_defined?(:Record, false)
  Acta.send(:remove_const, :EventsRecord)  if Acta.const_defined?(:EventsRecord, false)

  Acta.const_set(:EventsRecord, Class.new(parent) { self.abstract_class = true })
  Acta.const_set(:Record, Class.new(Acta::EventsRecord) do
    self.table_name = "events"
    self.inheritance_column = nil
  end)
end

.subscribe(event_class, handler_class, &block) ⇒ Object 



69
70
71
 
# File 'lib/acta.rb', line 69

def self.subscribe(event_class, handler_class, &block)
  handlers[event_class] << { handler_class:, block:, kind: handler_kind(handler_class) }
end

.upcaster_registry ⇒ Object 



166
167
168
 
# File 'lib/acta.rb', line 166

def self.upcaster_registry
  @upcaster_registry ||= Upcaster::Registry.new
end

.version_of(stream_type:, stream_key:) ⇒ Object

Public: read the current high-water mark for a stream. Returns 0 for streams that have never been emitted to. Use the result with ‘Acta.emit(…, if_version: version)` for optimistic locking.



279
280
281
282
283
 
# File 'lib/acta.rb', line 279

def self.version_of(stream_type:, stream_key:)
  Record
    .where(stream_type: stream_type.to_s, stream_key: stream_key)
    .maximum(:stream_sequence) || 0
end