Sha256: f7eb12a8c734c6f1fc4c7c6660998dfcdbf1da27550cdb3d3966b5d489ffadf2
Contents?: true
Size: 1.43 KB
Versions: 7
Compression:
Stored size: 1.43 KB
Contents
module SandthornDriverSequel
class EventQuery
def initialize(storage)
@storage = storage
end
def build(
aggregate_types: [],
take: 0,
after_sequence_number: 0)
aggregate_types.map!(&:to_s)
query = storage.events
query = add_aggregate_types(query, aggregate_types)
query = add_sequence_number(query, after_sequence_number)
query = add_select(query)
query = add_limit(query, take)
@query = query.order(:sequence_number)
end
def events
@query.all
end
private
attr_reader :storage
def add_limit(query, take)
if take > 0
query.limit(take)
else
query
end
end
def add_select(query)
query.select(*select_columns)
end
def select_columns
rel = Sequel.qualify(storage.events_table_name, :aggregate_version)
[
:aggregate_type,
rel,
:aggregate_id,
:sequence_number,
:event_name,
:event_data,
:timestamp
]
end
def add_sequence_number(query, after_sequence_number)
query.where { sequence_number > after_sequence_number }
end
def add_aggregate_types(query, aggregate_types)
if aggregate_types.any?
query.join(storage.aggregates, id: :aggregate_table_id, aggregate_type: aggregate_types)
else
query.join(storage.aggregates, id: :aggregate_table_id)
end
end
end
end
Version data entries
7 entries across 7 versions & 1 rubygems