mirror of
https://github.com/andwati/zitadel.git
synced 2026-11-10 16:15:12 +00:00
Eventstore (#22)
* feat: eventstore repository * fix: remove gorm * version * feat: pkg * feat: eventstore without eventstore-lib * rename files * gnueg * fix: add object * fix: global model * feat(eventstore): sdk * fix(eventstore): search query * fix(eventstore): rename app to eventstore * delete empty test * remove unused func * merge master * fix(eventstore): tests * fix(models): delete unused struct * feat(eventstore): implemented push events * feat(eventstore): overwrite context data * feat(types): SQL-config * feat(eventstore): options to overwrite editor * Update internal/eventstore/models/field.go Co-Authored-By: livio-a <[email protected]> * fix(eventstore): code quality * fix(eventstore): rename modifier* to editor* * fix(eventstore): delete editor_org Co-authored-by: Fabiennne <[email protected]> Co-authored-by: livio-a <[email protected]>
This commit is contained in:
Silvan
co-authored by
livio-a
Fabiennne
livio-a
Fabiennne
parent
d6e97ff1fc
commit
fbeab4c582
@@ -0,0 +1,24 @@
|
||||
package sql
|
||||
|
||||
import (
|
||||
// postgres dialect
|
||||
"database/sql"
|
||||
|
||||
"github.com/caos/zitadel/internal/config/types"
|
||||
"github.com/caos/zitadel/internal/errors"
|
||||
_ "github.com/lib/pq"
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
SQL types.SQL
|
||||
}
|
||||
|
||||
func Start(conf Config) (*SQL, error) {
|
||||
client, err := sql.Open("postgres", conf.SQL.ConnectionString())
|
||||
if err != nil {
|
||||
return nil, errors.ThrowPreconditionFailed(err, "SQL-9qBtr", "unable to open database connection")
|
||||
}
|
||||
return &SQL{
|
||||
client: client,
|
||||
}, nil
|
||||
}
|
||||
@@ -0,0 +1,188 @@
|
||||
package sql
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/DATA-DOG/go-sqlmock"
|
||||
"github.com/caos/zitadel/internal/eventstore/models"
|
||||
"github.com/lib/pq"
|
||||
)
|
||||
|
||||
const (
|
||||
selectEscaped = `SELECT id, creation_date, event_type, event_sequence, previous_sequence, event_data, editor_service, editor_user, resource_owner, aggregate_type, aggregate_id, aggregate_version FROM eventstore\.events`
|
||||
)
|
||||
|
||||
var (
|
||||
expectedFilterEventsLimitFormat = regexp.MustCompile(selectEscaped + ` ORDER BY event_sequence LIMIT \$1`).String()
|
||||
expectedFilterEventsDescFormat = regexp.MustCompile(selectEscaped + ` ORDER BY event_sequence DESC`).String()
|
||||
expectedFilterEventsAggregateIDLimit = regexp.MustCompile(selectEscaped + ` WHERE aggregate_id = \$1 ORDER BY event_sequence LIMIT \$2`).String()
|
||||
expectedFilterEventsAggregateIDTypeLimit = regexp.MustCompile(selectEscaped + ` WHERE aggregate_id = \$1 AND aggregate_type IN \(\$2\) ORDER BY event_sequence LIMIT \$3`).String()
|
||||
expectedGetAllEvents = regexp.MustCompile(selectEscaped + ` ORDER BY event_sequence`).String()
|
||||
|
||||
expectedInsertStatement = regexp.MustCompile(`insert into eventstore\.events ` +
|
||||
`\(event_type, aggregate_type, aggregate_id, aggregate_version, creation_date, event_data, editor_user, editor_service, resource_owner, previous_sequence\) ` +
|
||||
`select \$1, \$2, \$3, \$4, coalesce\(\$5, now\(\)\), \$6, \$7, \$8, \$9, ` +
|
||||
`case \(select exists\(select event_sequence from eventstore\.events where aggregate_type = \$10 AND aggregate_id = \$11\)\) ` +
|
||||
`WHEN true then \(select event_sequence from eventstore\.events where aggregate_type = \$12 AND aggregate_id = \$13 order by event_sequence desc limit 1\) ` +
|
||||
`ELSE NULL ` +
|
||||
`end ` +
|
||||
`where \(` +
|
||||
`\(select count\(id\) from eventstore\.events where event_sequence >= \$14 AND aggregate_type = \$15 AND aggregate_id = \$16\) = 1 OR ` +
|
||||
`\(\(select count\(id\) from eventstore\.events where aggregate_type = \$17 and aggregate_id = \$18\) = 0 AND \$19 = 0\)\) RETURNING id, event_sequence, creation_date`).String()
|
||||
)
|
||||
|
||||
type dbMock struct {
|
||||
sqlClient *sql.DB
|
||||
mock sqlmock.Sqlmock
|
||||
}
|
||||
|
||||
func (db *dbMock) close() {
|
||||
db.sqlClient.Close()
|
||||
}
|
||||
|
||||
func mockDB(t *testing.T) *dbMock {
|
||||
mockDB := dbMock{}
|
||||
var err error
|
||||
mockDB.sqlClient, mockDB.mock, err = sqlmock.New()
|
||||
if err != nil {
|
||||
t.Fatalf("error occured while creating stub db %v", err)
|
||||
}
|
||||
|
||||
mockDB.mock.MatchExpectationsInOrder(true)
|
||||
|
||||
return &mockDB
|
||||
}
|
||||
|
||||
func (db *dbMock) expectBegin(err error) *dbMock {
|
||||
if err != nil {
|
||||
db.mock.ExpectBegin().WillReturnError(err)
|
||||
} else {
|
||||
db.mock.ExpectBegin()
|
||||
}
|
||||
return db
|
||||
}
|
||||
|
||||
func (db *dbMock) expectSavepoint() *dbMock {
|
||||
db.mock.ExpectExec("SAVEPOINT").WillReturnResult(sqlmock.NewResult(1, 1))
|
||||
|
||||
return db
|
||||
}
|
||||
|
||||
func (db *dbMock) expectReleaseSavepoint(err error) *dbMock {
|
||||
expectation := db.mock.ExpectExec("RELEASE SAVEPOINT")
|
||||
if err == nil {
|
||||
expectation.WillReturnResult(sqlmock.NewResult(1, 1))
|
||||
} else {
|
||||
expectation.WillReturnError(err)
|
||||
}
|
||||
|
||||
return db
|
||||
}
|
||||
|
||||
func (db *dbMock) expectCommit(err error) *dbMock {
|
||||
if err != nil {
|
||||
db.mock.ExpectCommit().WillReturnError(err)
|
||||
} else {
|
||||
db.mock.ExpectCommit()
|
||||
}
|
||||
return db
|
||||
}
|
||||
|
||||
func (db *dbMock) expectRollback(err error) *dbMock {
|
||||
if err != nil {
|
||||
db.mock.ExpectRollback().WillReturnError(err)
|
||||
} else {
|
||||
db.mock.ExpectRollback()
|
||||
}
|
||||
return db
|
||||
}
|
||||
|
||||
func (db *dbMock) expectInsertEvent(e *models.Event, returnedID string, returnedSequence uint64) *dbMock {
|
||||
db.mock.ExpectQuery(expectedInsertStatement).
|
||||
WithArgs(
|
||||
e.Type, e.AggregateType, e.AggregateID, e.AggregateVersion, sqlmock.AnyArg(), e.Data, e.EditorUser, e.EditorService, e.ResourceOwner,
|
||||
e.AggregateType, e.AggregateID,
|
||||
e.AggregateType, e.AggregateID,
|
||||
e.PreviousSequence, e.AggregateType, e.AggregateID,
|
||||
e.AggregateType, e.AggregateID, e.PreviousSequence,
|
||||
).
|
||||
WillReturnRows(
|
||||
sqlmock.NewRows([]string{"id", "event_sequence", "creation_date"}).
|
||||
AddRow(returnedID, returnedSequence, time.Now().UTC()),
|
||||
)
|
||||
|
||||
return db
|
||||
}
|
||||
|
||||
func (db *dbMock) expectInsertEventError(e *models.Event) *dbMock {
|
||||
db.mock.ExpectQuery(expectedInsertStatement).
|
||||
WithArgs(
|
||||
e.Type, e.AggregateType, e.AggregateID, e.AggregateVersion, sqlmock.AnyArg(), e.Data, e.EditorUser, e.EditorService, e.ResourceOwner,
|
||||
e.AggregateType, e.AggregateID,
|
||||
e.AggregateType, e.AggregateID,
|
||||
e.PreviousSequence, e.AggregateType, e.AggregateID,
|
||||
e.AggregateType, e.AggregateID, e.PreviousSequence,
|
||||
).
|
||||
WillReturnError(sql.ErrTxDone)
|
||||
|
||||
return db
|
||||
}
|
||||
|
||||
func (db *dbMock) expectFilterEventsLimit(limit uint64, eventCount int) *dbMock {
|
||||
rows := sqlmock.NewRows([]string{"id", "creation_date"})
|
||||
for i := 0; i < eventCount; i++ {
|
||||
rows.AddRow(fmt.Sprint("event", i), time.Now())
|
||||
}
|
||||
db.mock.ExpectQuery(expectedFilterEventsLimitFormat).
|
||||
WithArgs(limit).
|
||||
WillReturnRows(rows)
|
||||
return db
|
||||
}
|
||||
|
||||
func (db *dbMock) expectFilterEventsDesc(eventCount int) *dbMock {
|
||||
rows := sqlmock.NewRows([]string{"id", "creation_date"})
|
||||
for i := eventCount; i > 0; i-- {
|
||||
rows.AddRow(fmt.Sprint("event", i), time.Now())
|
||||
}
|
||||
db.mock.ExpectQuery(expectedFilterEventsDescFormat).
|
||||
WillReturnRows(rows)
|
||||
return db
|
||||
}
|
||||
|
||||
func (db *dbMock) expectFilterEventsAggregateIDLimit(aggregateID string, limit uint64) *dbMock {
|
||||
rows := sqlmock.NewRows([]string{"id", "creation_date"})
|
||||
for i := limit; i > 0; i-- {
|
||||
rows.AddRow(fmt.Sprint("event", i), time.Now())
|
||||
}
|
||||
db.mock.ExpectQuery(expectedFilterEventsAggregateIDLimit).
|
||||
WithArgs(aggregateID, limit).
|
||||
WillReturnRows(rows)
|
||||
return db
|
||||
}
|
||||
|
||||
func (db *dbMock) expectFilterEventsAggregateIDTypeLimit(aggregateID, aggregateType string, limit uint64) *dbMock {
|
||||
rows := sqlmock.NewRows([]string{"id", "creation_date"})
|
||||
for i := limit; i > 0; i-- {
|
||||
rows.AddRow(fmt.Sprint("event", i), time.Now())
|
||||
}
|
||||
db.mock.ExpectQuery(expectedFilterEventsAggregateIDTypeLimit).
|
||||
WithArgs(aggregateID, pq.Array([]string{aggregateType}), limit).
|
||||
WillReturnRows(rows)
|
||||
return db
|
||||
}
|
||||
|
||||
func (db *dbMock) expectFilterEventsError(returnedErr error) *dbMock {
|
||||
db.mock.ExpectQuery(expectedGetAllEvents).
|
||||
WillReturnError(returnedErr)
|
||||
return db
|
||||
}
|
||||
|
||||
func (db *dbMock) expectPrepareInsert() *dbMock {
|
||||
db.mock.ExpectPrepare(expectedInsertStatement)
|
||||
|
||||
return db
|
||||
}
|
||||
@@ -0,0 +1,156 @@
|
||||
package sql
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/caos/logging"
|
||||
"github.com/caos/zitadel/internal/errors"
|
||||
"github.com/caos/zitadel/internal/eventstore/models"
|
||||
es_models "github.com/caos/zitadel/internal/eventstore/models"
|
||||
"github.com/lib/pq"
|
||||
)
|
||||
|
||||
const (
|
||||
selectStmt = "SELECT" +
|
||||
" id" +
|
||||
", creation_date" +
|
||||
", event_type" +
|
||||
", event_sequence" +
|
||||
", previous_sequence" +
|
||||
", event_data" +
|
||||
", editor_service" +
|
||||
", editor_user" +
|
||||
", resource_owner" +
|
||||
", aggregate_type" +
|
||||
", aggregate_id" +
|
||||
", aggregate_version" +
|
||||
" FROM eventstore.events"
|
||||
)
|
||||
|
||||
func (db *SQL) Filter(ctx context.Context, searchQuery *es_models.SearchQuery) (events []*models.Event, err error) {
|
||||
where, values := prepareWhere(searchQuery)
|
||||
query := selectStmt + where
|
||||
|
||||
query += " ORDER BY event_sequence"
|
||||
if searchQuery.Desc {
|
||||
query += " DESC"
|
||||
}
|
||||
|
||||
if searchQuery.Limit > 0 {
|
||||
values = append(values, searchQuery.Limit)
|
||||
query += " LIMIT ?"
|
||||
}
|
||||
|
||||
query = numberPlaceholder(query, "?", "$")
|
||||
|
||||
rows, err := db.client.Query(query, values...)
|
||||
if err != nil {
|
||||
logging.Log("SQL-HP3Uk").WithError(err).Info("query failed")
|
||||
return nil, errors.ThrowInternal(err, "SQL-IJuyR", "unable to filter events")
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
events = make([]*es_models.Event, 0, searchQuery.Limit)
|
||||
|
||||
for rows.Next() {
|
||||
event := new(models.Event)
|
||||
events = append(events, event)
|
||||
|
||||
rows.Scan(
|
||||
&event.ID,
|
||||
&event.CreationDate,
|
||||
&event.Type,
|
||||
&event.Sequence,
|
||||
&event.PreviousSequence,
|
||||
&event.Data,
|
||||
&event.EditorService,
|
||||
&event.EditorUser,
|
||||
&event.ResourceOwner,
|
||||
&event.AggregateType,
|
||||
&event.AggregateID,
|
||||
&event.AggregateVersion,
|
||||
)
|
||||
}
|
||||
|
||||
return events, nil
|
||||
}
|
||||
|
||||
func numberPlaceholder(query, old, new string) string {
|
||||
for i, hasChanged := 1, true; hasChanged; i++ {
|
||||
newQuery := strings.Replace(query, old, new+strconv.Itoa(i), 1)
|
||||
hasChanged = query != newQuery
|
||||
query = newQuery
|
||||
}
|
||||
return query
|
||||
}
|
||||
|
||||
func prepareWhere(searchQuery *es_models.SearchQuery) (clause string, values []interface{}) {
|
||||
values = make([]interface{}, len(searchQuery.Filters))
|
||||
clauses := make([]string, len(searchQuery.Filters))
|
||||
|
||||
if len(values) == 0 {
|
||||
return clause, values
|
||||
}
|
||||
|
||||
for i, filter := range searchQuery.Filters {
|
||||
value := filter.GetValue()
|
||||
switch value.(type) {
|
||||
case []bool, []float64, []int64, []string, *[]bool, *[]float64, *[]int64, *[]string:
|
||||
value = pq.Array(value)
|
||||
}
|
||||
|
||||
clauses[i] = getCondition(filter)
|
||||
values[i] = value
|
||||
}
|
||||
return " WHERE " + strings.Join(clauses, " AND "), values
|
||||
}
|
||||
|
||||
func getCondition(filter *es_models.Filter) string {
|
||||
field := getField(filter.GetField())
|
||||
operation := getOperation(filter.GetOperation())
|
||||
format := prepareConditionFormat(filter.GetOperation())
|
||||
|
||||
return fmt.Sprintf(format, field, operation)
|
||||
}
|
||||
|
||||
func prepareConditionFormat(operation es_models.Operation) string {
|
||||
if operation == es_models.Operation_In {
|
||||
return "%s %s (?)"
|
||||
}
|
||||
return "%s %s ?"
|
||||
}
|
||||
|
||||
func getField(field es_models.Field) string {
|
||||
switch field {
|
||||
case es_models.Field_AggregateID:
|
||||
return "aggregate_id"
|
||||
case es_models.Field_AggregateType:
|
||||
return "aggregate_type"
|
||||
case es_models.Field_LatestSequence:
|
||||
return "event_sequence"
|
||||
case es_models.Field_ResourceOwner:
|
||||
return "resource_owner"
|
||||
case es_models.Field_ModifierService:
|
||||
return "editor_service"
|
||||
case es_models.Field_ModifierUser:
|
||||
return "editor_user"
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func getOperation(operation es_models.Operation) string {
|
||||
switch operation {
|
||||
case es_models.Operation_Equals:
|
||||
return "="
|
||||
case es_models.Operation_Greater:
|
||||
return ">"
|
||||
case es_models.Operation_Less:
|
||||
return "<"
|
||||
case es_models.Operation_In:
|
||||
return "IN"
|
||||
}
|
||||
return ""
|
||||
}
|
||||
@@ -0,0 +1,155 @@
|
||||
package sql
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"testing"
|
||||
|
||||
"github.com/caos/zitadel/internal/errors"
|
||||
es_models "github.com/caos/zitadel/internal/eventstore/models"
|
||||
)
|
||||
|
||||
func TestSQL_Filter(t *testing.T) {
|
||||
type fields struct {
|
||||
client *dbMock
|
||||
}
|
||||
type args struct {
|
||||
events *mockEvents
|
||||
searchQuery *es_models.SearchQuery
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
fields fields
|
||||
args args
|
||||
eventsLen int
|
||||
wantErr bool
|
||||
isErrFunc func(error) bool
|
||||
}{
|
||||
{
|
||||
name: "only limit filter",
|
||||
fields: fields{
|
||||
client: mockDB(t).expectFilterEventsLimit(34, 3),
|
||||
},
|
||||
args: args{
|
||||
events: &mockEvents{t: t},
|
||||
searchQuery: es_models.NewSearchQuery().SetLimit(34),
|
||||
},
|
||||
eventsLen: 3,
|
||||
wantErr: false,
|
||||
},
|
||||
{
|
||||
name: "only desc filter",
|
||||
fields: fields{
|
||||
client: mockDB(t).expectFilterEventsDesc(34),
|
||||
},
|
||||
args: args{
|
||||
events: &mockEvents{t: t},
|
||||
searchQuery: es_models.NewSearchQuery().OrderDesc(),
|
||||
},
|
||||
eventsLen: 34,
|
||||
wantErr: false,
|
||||
},
|
||||
{
|
||||
name: "no events found",
|
||||
fields: fields{
|
||||
client: mockDB(t).expectFilterEventsError(sql.ErrNoRows),
|
||||
},
|
||||
args: args{
|
||||
events: &mockEvents{t: t},
|
||||
searchQuery: &es_models.SearchQuery{},
|
||||
},
|
||||
wantErr: true,
|
||||
isErrFunc: errors.IsInternal,
|
||||
},
|
||||
{
|
||||
name: "filter fails because sql internal error",
|
||||
fields: fields{
|
||||
client: mockDB(t).expectFilterEventsError(sql.ErrConnDone),
|
||||
},
|
||||
args: args{
|
||||
events: &mockEvents{t: t},
|
||||
searchQuery: &es_models.SearchQuery{},
|
||||
},
|
||||
wantErr: true,
|
||||
isErrFunc: errors.IsInternal,
|
||||
},
|
||||
{
|
||||
name: "filter by aggregate id",
|
||||
fields: fields{
|
||||
client: mockDB(t).expectFilterEventsAggregateIDLimit("hop", 5),
|
||||
},
|
||||
args: args{
|
||||
events: &mockEvents{t: t},
|
||||
searchQuery: es_models.NewSearchQuery().SetLimit(5).AggregateIDFilter("hop"),
|
||||
},
|
||||
wantErr: false,
|
||||
isErrFunc: nil,
|
||||
},
|
||||
{
|
||||
name: "filter by aggregate id and aggregate type",
|
||||
fields: fields{
|
||||
client: mockDB(t).expectFilterEventsAggregateIDTypeLimit("hop", "user", 5),
|
||||
},
|
||||
args: args{
|
||||
events: &mockEvents{t: t},
|
||||
searchQuery: es_models.NewSearchQuery().SetLimit(5).AggregateIDFilter("hop").AggregateTypeFilter("user"),
|
||||
},
|
||||
wantErr: false,
|
||||
isErrFunc: nil,
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
sql := &SQL{
|
||||
client: tt.fields.client.sqlClient,
|
||||
}
|
||||
events, err := sql.Filter(context.Background(), tt.args.searchQuery)
|
||||
if (err != nil) != tt.wantErr {
|
||||
t.Errorf("SQL.UnlockAggregates() error = %v, wantErr %v", err, tt.wantErr)
|
||||
}
|
||||
if tt.eventsLen != 0 && len(events) != tt.eventsLen {
|
||||
t.Errorf("events has wrong length got: %d want %d", len(events), tt.eventsLen)
|
||||
}
|
||||
if tt.wantErr && !tt.isErrFunc(err) {
|
||||
t.Errorf("got wrong error %v", err)
|
||||
}
|
||||
if err := tt.fields.client.mock.ExpectationsWereMet(); err != nil {
|
||||
t.Errorf("there were unfulfilled expectations: %s", err)
|
||||
}
|
||||
tt.fields.client.close()
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func Test_getCondition(t *testing.T) {
|
||||
type args struct {
|
||||
filter *es_models.Filter
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
args args
|
||||
want string
|
||||
}{
|
||||
{
|
||||
name: "single value",
|
||||
args: args{
|
||||
filter: es_models.NewFilter(es_models.Field_LatestSequence, 34, es_models.Operation_Greater),
|
||||
},
|
||||
want: "event_sequence > ?",
|
||||
},
|
||||
{
|
||||
name: "list value",
|
||||
args: args{
|
||||
filter: es_models.NewFilter(es_models.Field_AggregateType, []string{"a", "b"}, es_models.Operation_In),
|
||||
},
|
||||
want: "aggregate_type IN (?)",
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
if got := getCondition(tt.args.filter); got != tt.want {
|
||||
t.Errorf("getCondition() = %v, want %v", got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,91 @@
|
||||
package sql
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
|
||||
"github.com/caos/logging"
|
||||
"github.com/caos/zitadel/internal/errors"
|
||||
"github.com/caos/zitadel/internal/eventstore/models"
|
||||
"github.com/cockroachdb/cockroach-go/crdb"
|
||||
)
|
||||
|
||||
const insertStmt = "insert into eventstore.events " +
|
||||
"(event_type, aggregate_type, aggregate_id, aggregate_version, creation_date, event_data, editor_user, editor_service, resource_owner, previous_sequence) " +
|
||||
"select $1, $2, $3, $4, coalesce($5, now()), $6, $7, $8, $9, " +
|
||||
// case is to set the highest sequence or NULL in previous_sequence
|
||||
"case (select exists(select event_sequence from eventstore.events where aggregate_type = $10 AND aggregate_id = $11)) " +
|
||||
"WHEN true then (select event_sequence from eventstore.events where aggregate_type = $12 AND aggregate_id = $13 order by event_sequence desc limit 1) " +
|
||||
"ELSE NULL " +
|
||||
"end " +
|
||||
"where (" +
|
||||
// exactly one event of requested aggregate must have a >= sequence (last inserted event)
|
||||
"(select count(id) from eventstore.events where event_sequence >= $14 AND aggregate_type = $15 AND aggregate_id = $16) = 1 OR " +
|
||||
// previous sequence = 0, no events must exist for the requested aggregate
|
||||
"((select count(id) from eventstore.events where aggregate_type = $17 and aggregate_id = $18) = 0 AND $19 = 0)) " +
|
||||
"RETURNING id, event_sequence, creation_date"
|
||||
|
||||
func (db *SQL) PushAggregates(ctx context.Context, aggregates ...*models.Aggregate) (err error) {
|
||||
err = crdb.ExecuteTx(ctx, db.client, nil, func(tx *sql.Tx) error {
|
||||
stmt, err := tx.Prepare(insertStmt)
|
||||
if err != nil {
|
||||
tx.Rollback()
|
||||
logging.Log("SQL-9ctx5").WithError(err).Warn("prepare failed")
|
||||
return errors.ThrowInternal(err, "SQL-juCgA", "prepare failed")
|
||||
}
|
||||
|
||||
for _, aggregate := range aggregates {
|
||||
err = insertEvents(stmt, aggregate.Events)
|
||||
if err != nil {
|
||||
tx.Rollback()
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
|
||||
if _, ok := err.(*errors.CaosError); !ok && err != nil {
|
||||
err = errors.ThrowInternal(err, "SQL-DjgtG", "unable to store events")
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
func insertEvents(stmt *sql.Stmt, events []*models.Event) error {
|
||||
previousSequence := events[0].PreviousSequence
|
||||
for _, event := range events {
|
||||
event.PreviousSequence = previousSequence
|
||||
|
||||
if event.Data == nil || len(event.Data) == 0 {
|
||||
//json decoder failes with EOF if json text is empty
|
||||
event.Data = []byte("{}")
|
||||
}
|
||||
|
||||
rows, err := stmt.Query(event.Type, event.AggregateType, event.AggregateID, event.AggregateVersion, event.CreationDate, event.Data, event.EditorUser, event.EditorService, event.ResourceOwner,
|
||||
event.AggregateType, event.AggregateID,
|
||||
event.AggregateType, event.AggregateID,
|
||||
event.PreviousSequence, event.AggregateType, event.AggregateID,
|
||||
event.AggregateType, event.AggregateID, event.PreviousSequence)
|
||||
|
||||
if err != nil {
|
||||
logging.Log("SQL-EXA0q").WithError(err).Info("query failed")
|
||||
return errors.ThrowInternal(err, "SQL-SBP37", "unable to create event")
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
rowInserted := false
|
||||
for rows.Next() {
|
||||
rowInserted = true
|
||||
err = rows.Scan(&event.ID, &event.Sequence, &event.CreationDate)
|
||||
logging.Log("SQL-rAvLD").OnError(err).Info("unable to scan result into event")
|
||||
}
|
||||
|
||||
if !rowInserted {
|
||||
return errors.ThrowAlreadyExists(nil, "SQL-GKcAa", "wrong sequence")
|
||||
}
|
||||
|
||||
previousSequence = event.Sequence
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,323 @@
|
||||
package sql
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"reflect"
|
||||
"runtime"
|
||||
"testing"
|
||||
|
||||
"github.com/caos/zitadel/internal/errors"
|
||||
"github.com/caos/zitadel/internal/eventstore/models"
|
||||
)
|
||||
|
||||
type mockEvents struct {
|
||||
events []*models.Event
|
||||
t *testing.T
|
||||
}
|
||||
|
||||
func TestSQL_PushAggregates(t *testing.T) {
|
||||
type fields struct {
|
||||
client *dbMock
|
||||
}
|
||||
type args struct {
|
||||
aggregates []*models.Aggregate
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
fields fields
|
||||
args args
|
||||
isError func(error) bool
|
||||
shouldCheckEvents bool
|
||||
}{
|
||||
{
|
||||
name: "no aggregates",
|
||||
fields: fields{
|
||||
client: mockDB(t).
|
||||
expectBegin(nil).
|
||||
expectSavepoint().
|
||||
expectPrepareInsert().
|
||||
expectReleaseSavepoint(nil).
|
||||
expectCommit(nil),
|
||||
},
|
||||
args: args{aggregates: []*models.Aggregate{}},
|
||||
shouldCheckEvents: false,
|
||||
isError: noErr,
|
||||
},
|
||||
{
|
||||
name: "no aggregates release fails",
|
||||
fields: fields{
|
||||
client: mockDB(t).
|
||||
expectBegin(nil).
|
||||
expectSavepoint().
|
||||
expectPrepareInsert().
|
||||
expectReleaseSavepoint(sql.ErrConnDone).
|
||||
expectCommit(nil),
|
||||
},
|
||||
|
||||
args: args{aggregates: []*models.Aggregate{}},
|
||||
isError: errors.IsInternal,
|
||||
shouldCheckEvents: false,
|
||||
},
|
||||
{
|
||||
name: "one aggregate two events success",
|
||||
fields: fields{
|
||||
client: mockDB(t).
|
||||
expectBegin(nil).
|
||||
expectSavepoint().
|
||||
expectPrepareInsert().
|
||||
expectInsertEvent(&models.Event{
|
||||
AggregateID: "aggID",
|
||||
AggregateType: "aggType",
|
||||
EditorService: "svc",
|
||||
EditorUser: "usr",
|
||||
ResourceOwner: "ro",
|
||||
PreviousSequence: 34,
|
||||
Type: "eventTyp",
|
||||
Data: []byte("{}"),
|
||||
AggregateVersion: "v0.0.1",
|
||||
},
|
||||
"asdfölk-234", 45).
|
||||
expectInsertEvent(&models.Event{
|
||||
AggregateID: "aggID",
|
||||
AggregateType: "aggType",
|
||||
EditorService: "svc2",
|
||||
EditorUser: "usr2",
|
||||
ResourceOwner: "ro2",
|
||||
PreviousSequence: 45,
|
||||
Type: "eventTyp",
|
||||
Data: []byte("{}"),
|
||||
AggregateVersion: "v0.0.1",
|
||||
}, "asdfölk-233", 46).
|
||||
expectReleaseSavepoint(nil).
|
||||
expectCommit(nil),
|
||||
},
|
||||
args: args{
|
||||
aggregates: []*models.Aggregate{
|
||||
&models.Aggregate{
|
||||
Events: []*models.Event{
|
||||
&models.Event{
|
||||
AggregateID: "aggID",
|
||||
AggregateType: "aggType",
|
||||
AggregateVersion: "v0.0.1",
|
||||
EditorService: "svc",
|
||||
EditorUser: "usr",
|
||||
ResourceOwner: "ro",
|
||||
Type: "eventTyp",
|
||||
PreviousSequence: 34,
|
||||
},
|
||||
&models.Event{
|
||||
AggregateID: "aggID",
|
||||
AggregateType: "aggType",
|
||||
AggregateVersion: "v0.0.1",
|
||||
EditorService: "svc2",
|
||||
EditorUser: "usr2",
|
||||
ResourceOwner: "ro2",
|
||||
Type: "eventTyp",
|
||||
PreviousSequence: 0,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
shouldCheckEvents: true,
|
||||
isError: noErr,
|
||||
},
|
||||
{
|
||||
name: "two aggregates one event per aggregate success",
|
||||
fields: fields{
|
||||
client: mockDB(t).
|
||||
expectBegin(nil).
|
||||
expectSavepoint().
|
||||
expectPrepareInsert().
|
||||
expectInsertEvent(&models.Event{
|
||||
AggregateID: "aggID",
|
||||
AggregateType: "aggType",
|
||||
EditorService: "svc",
|
||||
EditorUser: "usr",
|
||||
ResourceOwner: "ro",
|
||||
PreviousSequence: 34,
|
||||
Data: []byte("{}"),
|
||||
Type: "eventTyp",
|
||||
AggregateVersion: "v0.0.1",
|
||||
}, "asdfölk-233", 47).
|
||||
expectInsertEvent(&models.Event{
|
||||
AggregateID: "aggID2",
|
||||
AggregateType: "aggType2",
|
||||
EditorService: "svc",
|
||||
EditorUser: "usr",
|
||||
ResourceOwner: "ro",
|
||||
PreviousSequence: 40,
|
||||
Data: []byte("{}"),
|
||||
Type: "eventTyp",
|
||||
AggregateVersion: "v0.0.1",
|
||||
}, "asdfölk-233", 48).
|
||||
expectReleaseSavepoint(nil).
|
||||
expectCommit(nil),
|
||||
},
|
||||
args: args{
|
||||
aggregates: []*models.Aggregate{
|
||||
&models.Aggregate{
|
||||
Events: []*models.Event{
|
||||
&models.Event{
|
||||
AggregateID: "aggID",
|
||||
AggregateType: "aggType",
|
||||
AggregateVersion: "v0.0.1",
|
||||
EditorService: "svc",
|
||||
EditorUser: "usr",
|
||||
ResourceOwner: "ro",
|
||||
Type: "eventTyp",
|
||||
PreviousSequence: 34,
|
||||
},
|
||||
},
|
||||
},
|
||||
&models.Aggregate{
|
||||
Events: []*models.Event{
|
||||
&models.Event{
|
||||
AggregateID: "aggID2",
|
||||
AggregateType: "aggType2",
|
||||
AggregateVersion: "v0.0.1",
|
||||
EditorService: "svc",
|
||||
EditorUser: "usr",
|
||||
ResourceOwner: "ro",
|
||||
Type: "eventTyp",
|
||||
PreviousSequence: 40,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
shouldCheckEvents: true,
|
||||
isError: noErr,
|
||||
},
|
||||
{
|
||||
name: "first event fails no action with second event",
|
||||
fields: fields{
|
||||
client: mockDB(t).
|
||||
expectBegin(nil).
|
||||
expectSavepoint().
|
||||
expectInsertEventError(&models.Event{
|
||||
AggregateID: "aggID",
|
||||
AggregateType: "aggType",
|
||||
EditorService: "svc",
|
||||
EditorUser: "usr",
|
||||
ResourceOwner: "ro",
|
||||
PreviousSequence: 34,
|
||||
Data: []byte("{}"),
|
||||
Type: "eventTyp",
|
||||
AggregateVersion: "v0.0.1",
|
||||
}).
|
||||
expectReleaseSavepoint(nil).
|
||||
expectRollback(nil),
|
||||
},
|
||||
args: args{
|
||||
aggregates: []*models.Aggregate{
|
||||
&models.Aggregate{
|
||||
Events: []*models.Event{
|
||||
&models.Event{
|
||||
AggregateID: "aggID",
|
||||
AggregateType: "aggType",
|
||||
AggregateVersion: "v0.0.1",
|
||||
EditorService: "svc",
|
||||
EditorUser: "usr",
|
||||
ResourceOwner: "ro",
|
||||
Type: "eventTyp",
|
||||
PreviousSequence: 34,
|
||||
},
|
||||
&models.Event{
|
||||
AggregateID: "aggID",
|
||||
AggregateType: "aggType",
|
||||
AggregateVersion: "v0.0.1",
|
||||
EditorService: "svc",
|
||||
EditorUser: "usr",
|
||||
ResourceOwner: "ro",
|
||||
Type: "eventTyp",
|
||||
PreviousSequence: 0,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
isError: errors.IsInternal,
|
||||
shouldCheckEvents: false,
|
||||
},
|
||||
{
|
||||
name: "one event, release savepoint fails",
|
||||
fields: fields{
|
||||
client: mockDB(t).
|
||||
expectBegin(nil).
|
||||
expectPrepareInsert().
|
||||
expectSavepoint().
|
||||
expectInsertEvent(&models.Event{
|
||||
AggregateID: "aggID",
|
||||
AggregateType: "aggType",
|
||||
EditorService: "svc",
|
||||
EditorUser: "usr",
|
||||
ResourceOwner: "ro",
|
||||
PreviousSequence: 34,
|
||||
Type: "eventTyp",
|
||||
Data: []byte("{}"),
|
||||
AggregateVersion: "v0.0.1",
|
||||
}, "asdfölk-233", 47).
|
||||
expectReleaseSavepoint(sql.ErrConnDone).
|
||||
expectCommit(nil).
|
||||
expectRollback(nil),
|
||||
},
|
||||
args: args{
|
||||
aggregates: []*models.Aggregate{
|
||||
&models.Aggregate{
|
||||
Events: []*models.Event{
|
||||
&models.Event{
|
||||
AggregateID: "aggID",
|
||||
AggregateType: "aggType",
|
||||
AggregateVersion: "v0.0.1",
|
||||
EditorService: "svc",
|
||||
EditorUser: "usr",
|
||||
ResourceOwner: "ro",
|
||||
Type: "eventTyp",
|
||||
PreviousSequence: 34,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
isError: errors.IsInternal,
|
||||
shouldCheckEvents: false,
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
sql := &SQL{
|
||||
client: tt.fields.client.sqlClient,
|
||||
}
|
||||
err := sql.PushAggregates(context.Background(), tt.args.aggregates...)
|
||||
if err != nil && !tt.isError(err) {
|
||||
t.Errorf("wrong error type = %v, errFunc %s", err, functionName(tt.isError))
|
||||
}
|
||||
if !tt.shouldCheckEvents {
|
||||
return
|
||||
}
|
||||
for _, aggregate := range tt.args.aggregates {
|
||||
for _, event := range aggregate.Events {
|
||||
if event.Sequence == 0 {
|
||||
t.Error("sequence of returned event is not set")
|
||||
}
|
||||
if event.AggregateType == "" || event.AggregateID == "" {
|
||||
t.Error("aggregate of event is not set")
|
||||
}
|
||||
}
|
||||
}
|
||||
if err := tt.fields.client.mock.ExpectationsWereMet(); err != nil {
|
||||
t.Errorf("not all database expectaions met: %s", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func noErr(err error) bool {
|
||||
return err == nil
|
||||
}
|
||||
|
||||
func functionName(i interface{}) string {
|
||||
return runtime.FuncForPC(reflect.ValueOf(i).Pointer()).Name()
|
||||
}
|
||||
@@ -0,0 +1,17 @@
|
||||
package sql
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
|
||||
//sql import
|
||||
_ "github.com/lib/pq"
|
||||
)
|
||||
|
||||
type SQL struct {
|
||||
client *sql.DB
|
||||
}
|
||||
|
||||
func (db *SQL) Health(ctx context.Context) error {
|
||||
return db.client.Ping()
|
||||
}
|
||||
Reference in New Issue
Block a user