From 46a68c15bfc4342b6f4bad42dfa23187e8684c00 Mon Sep 17 00:00:00 2001 From: adlerhurst Date: Tue, 6 Oct 2020 20:20:23 +0200 Subject: [PATCH] fix: eventstore v2 insert statement --- internal/eventstore/v2/repository/sql/crdb.go | 38 +++-- .../v2/repository/sql/local_crdb_test.go | 137 +++++++++++++----- 2 files changed, 117 insertions(+), 58 deletions(-) diff --git a/internal/eventstore/v2/repository/sql/crdb.go b/internal/eventstore/v2/repository/sql/crdb.go index ea67e7651..1f0d88cc1 100644 --- a/internal/eventstore/v2/repository/sql/crdb.go +++ b/internal/eventstore/v2/repository/sql/crdb.go @@ -30,8 +30,7 @@ const ( " previous_sequence, " + " check_previous, " + // variables below are calculated - " max_event_seq, " + - " event_count " + + " max_event_seq " + ") " + " AS( " + " SELECT " + @@ -46,8 +45,7 @@ const ( " $9::VARCHAR, " + " $10::BIGINT, " + " $11::BOOLEAN," + - " MAX(event_sequence) AS max_event_seq, " + - " COUNT(*) AS event_count " + + " MAX(event_sequence) AS max_event_seq " + " FROM eventstore.events " + " WHERE " + " aggregate_type = $2::VARCHAR " + @@ -80,25 +78,23 @@ const ( " ( " + " SELECT " + " CASE " + - " WHEN NOT check_previous THEN " + - " max_event_seq " + - " ELSE " + - " previous_sequence " + + " WHEN NOT check_previous " + + " THEN max_event_seq " + + " ELSE previous_sequence " + " END" + " ) " + " FROM input_event " + - " WHERE EXISTS ( " + - " SELECT " + + " WHERE 1 = " + " CASE " + - " WHEN NOT check_previous THEN 1 " + + " WHEN NOT check_previous " + + " THEN 1 " + " ELSE ( " + " SELECT 1 FROM input_event " + - " WHERE max_event_seq = previous_sequence OR (previous_sequence IS NULL AND event_count = 0) " + + " WHERE (max_event_seq IS NULL AND previous_sequence IS NULL) OR (max_event_seq IS NOT NULL AND max_event_seq = previous_sequence) " + " ) " + " END " + - " ) " + " ) " + - "RETURNING id, event_sequence, creation_date " + "RETURNING id, event_sequence, previous_sequence, creation_date " ) type CRDB struct { @@ -118,9 +114,9 @@ func (db *CRDB) Push(ctx context.Context, events ...*repository.Event) error { return caos_errs.ThrowInternal(err, "SQL-OdXRE", "prepare failed") } for _, event := range events { - previousSequence := event.PreviousSequence + previousSequence := Sequence(event.PreviousSequence) if event.PreviousEvent != nil { - previousSequence = event.PreviousSequence + previousSequence = Sequence(event.PreviousEvent.Sequence) } err = stmt.QueryRowContext(ctx, event.Type, @@ -135,9 +131,11 @@ func (db *CRDB) Push(ctx context.Context, events ...*repository.Event) error { event.EditorUser, event.EditorService, event.ResourceOwner, - Sequence(previousSequence), + previousSequence, event.CheckPreviousSequence, - ).Scan(&event.ID, &event.Sequence, &event.CreationDate) + ).Scan(&event.ID, &event.Sequence, &previousSequence, &event.CreationDate) + + event.PreviousSequence = uint64(previousSequence) if err != nil { tx.Rollback() @@ -180,7 +178,7 @@ func (db *CRDB) LatestSequence(ctx context.Context, searchQuery *repository.Sear return uint64(seq), nil } -func (db *CRDB) query(searchQuery *repository.SearchQuery, data interface{}) error { +func (db *CRDB) query(searchQuery *repository.SearchQuery, dest interface{}) error { query, values, rowScanner := buildQuery(db, searchQuery) if query == "" { return caos_errs.ThrowInvalidArgument(nil, "SQL-rWeBw", "invalid query factory") @@ -194,7 +192,7 @@ func (db *CRDB) query(searchQuery *repository.SearchQuery, data interface{}) err defer rows.Close() for rows.Next() { - err = rowScanner(rows.Scan, nil) + err = rowScanner(rows.Scan, dest) if err != nil { return err } diff --git a/internal/eventstore/v2/repository/sql/local_crdb_test.go b/internal/eventstore/v2/repository/sql/local_crdb_test.go index 49d666e18..93117d294 100644 --- a/internal/eventstore/v2/repository/sql/local_crdb_test.go +++ b/internal/eventstore/v2/repository/sql/local_crdb_test.go @@ -48,27 +48,29 @@ func TestMain(m *testing.M) { func TestInsert(t *testing.T) { crdb := &CRDB{client: db} + e1 := &repository.Event{ + AggregateID: "agg.id", + AggregateType: "agg.type", + CheckPreviousSequence: true, + EditorService: "edi.svc", + EditorUser: "edi", + ResourceOwner: "edit", + Type: "type", + Version: "v1", + } events := []*repository.Event{ + e1, { AggregateID: "agg.id", AggregateType: "agg.type", - CheckPreviousSequence: false, - EditorService: "edi.svc", - EditorUser: "edi", - ResourceOwner: "edit", - Type: "type", - Version: "v1", - }, - { - AggregateID: "agg.id", - AggregateType: "agg.type", - CheckPreviousSequence: false, + CheckPreviousSequence: true, EditorService: "edi.svc", EditorUser: "edi", ResourceOwner: "edit", Type: "type", Version: "v1", CreationDate: time.Now().Add(-2 * time.Second), + PreviousEvent: e1, }, { AggregateID: "agg.id", @@ -81,12 +83,66 @@ func TestInsert(t *testing.T) { Version: "v1", CreationDate: time.Now().Add(-500 * time.Millisecond), }, + { + AggregateID: "agg.id2", + AggregateType: "agg.type", + CheckPreviousSequence: true, + EditorService: "edi.svc", + EditorUser: "edi", + ResourceOwner: "edit", + Type: "type", + Version: "v1", + CreationDate: time.Now().Add(-500 * time.Millisecond), + }, + { + AggregateID: "agg.id3", + AggregateType: "agg.type", + CheckPreviousSequence: false, + EditorService: "edi.svc", + EditorUser: "edi", + ResourceOwner: "edit", + Type: "type", + Version: "v1", + CreationDate: time.Now().Add(-500 * time.Millisecond), + }, + { + AggregateID: "agg.id", + AggregateType: "agg.type", + CheckPreviousSequence: false, + EditorService: "edi.svc", + EditorUser: "edi", + ResourceOwner: "edit", + Type: "type", + Version: "v1", + CreationDate: time.Now().Add(-500 * time.Millisecond), + }, + // { + // AggregateID: "agg.id4", + // AggregateType: "agg.type", + // CheckPreviousSequence: false, + // EditorService: "edi.svc", + // EditorUser: "edi", + // ResourceOwner: "edit", + // Type: "type", + // Version: "v1", + // CreationDate: time.Now().Add(-500 * time.Millisecond), + // PreviousEvent: e1, + // }, + //fail because wrong previous event + // { + // AggregateID: "agg.id2", + // AggregateType: "agg.type", + // CheckPreviousSequence: true, + // EditorService: "edi.svc", + // EditorUser: "edi", + // ResourceOwner: "edit", + // Type: "type", + // Version: "v1", + // CreationDate: time.Now().Add(-500 * time.Millisecond), + // PreviousEvent: e1, + // }, } fmt.Println("==============") - fmt.Println("will insert ts:") - for _, event := range events { - fmt.Printf("%v | %v\n", event.CreationDate, event.ID) - } err := crdb.Push(context.Background(), events...) if err != nil { t.Error(err) @@ -95,35 +151,14 @@ func TestInsert(t *testing.T) { fmt.Println("inserted ts:") for _, event := range events { - fmt.Printf("%v | %v\n", event.CreationDate, event.ID) + fmt.Printf("%+v\n", event) } - fmt.Println("==============") - for _, event := range events { - fmt.Printf("%+v", event) - } - - // tx, _ := db.Begin() - - // var seq Sequence - // var d time.Time - - // row := tx.QueryRow(crdbInsert, "event.type", "aggregate.type", "aggregate.id", repository.Version("v1"), nil, Data(nil), "editor.user", "editor.service", "resource.owner", Sequence(0), false) - // err := row.Scan(&seq, &d) - - // row = tx.QueryRow(crdbInsert, "event.type", "aggregate.type", "aggregate.id", repository.Version("v1"), nil, Data(nil), "editor.user", "editor.service", "resource.owner", Sequence(1), true) - // err = row.Scan(&seq, &d) - - // row = tx.QueryRow(crdbInsert, "event.type", "aggregate.type", "aggregate.id", repository.Version("v1"), nil, Data(nil), "editor.user", "editor.service", "resource.owner", Sequence(0), false) - // err = row.Scan(&seq, &d) - - // tx.Commit() - + fmt.Println("====================") rows, err := db.Query("select * from eventstore.events order by event_sequence") defer rows.Close() fmt.Println(err) - fmt.Println(rows.Columns()) for rows.Next() { i := make([]interface{}, 12) var id string @@ -132,7 +167,33 @@ func TestInsert(t *testing.T) { fmt.Println(i) } + fmt.Println("====================") + filtededEvents, err := crdb.Filter(context.Background(), &repository.SearchQuery{ + Columns: repository.Columns_Event, + Filters: []*repository.Filter{ + { + Field: repository.Field_AggregateType, + Operation: repository.Operation_Equals, + Value: repository.AggregateType("agg.type"), + }, + }, + }) + fmt.Println(err) + for _, event := range filtededEvents { + fmt.Printf("%+v\n", event) + } + fmt.Println("====================") + rows, err = db.Query("select max(event_sequence), count(*) from eventstore.events where aggregate_type = 'agg.type' and aggregate_id = 'agg.id'") + defer rows.Close() + fmt.Println(err) + + for rows.Next() { + i := make([]interface{}, 2) + rows.Scan(&i[0], &i[1]) + + fmt.Println(i) + } t.Fail() }