feat: add assets to eventstore and event (#1674)

* fix: add assets to eventstore and event

* fix: project member, grant member, app changed tests

* fix: asset migrations

* feat: add asset tests

* feat: add asset tests

* Update internal/eventstore/repository/repository.go

Co-authored-by: Livio Amstutz <livio.a@gmail.com>

* feat: add asset tests

Co-authored-by: Livio Amstutz <livio.a@gmail.com>

internal/eventstore/event.go

@@ -20,6 +20,8 @@ type EventPusher interface {
 	// * struct which can be marshalled to json
 	// * pointer to struct which can be marshalled to json
 	Data() interface{}
+	//Assets contain assets in form of []byte, these will be stored to a separate table
+	Assets() []*Asset
 	//UniqueConstraints should be added for unique attributes of an event, if nil constraints will not be checked
 	UniqueConstraints() []*EventUniqueConstraint
 }

internal/eventstore/eventstore.go

@@ -39,11 +39,11 @@ func (es *Eventstore) Health(ctx context.Context) error {
 //PushEvents pushes the events in a single transaction
 // an event needs at least an aggregate
 func (es *Eventstore) PushEvents(ctx context.Context, pushEvents ...EventPusher) ([]EventReader, error) {
-	events, constraints, err := eventsToRepository(pushEvents)
+	events, assets, constraints, err := eventsToRepository(pushEvents)
 	if err != nil {
 		return nil, err
 	}
-	err = es.repo.Push(ctx, events, constraints...)
+	err = es.repo.Push(ctx, events, assets, constraints...)
 	if err != nil {
 		return nil, err
 	}
@@ -57,12 +57,12 @@ func (es *Eventstore) PushEvents(ctx context.Context, pushEvents ...EventPusher)
 	return eventReaders, nil
 }
 
-func eventsToRepository(pushEvents []EventPusher) (events []*repository.Event, constraints []*repository.UniqueConstraint, err error) {
+func eventsToRepository(pushEvents []EventPusher) (events []*repository.Event, assets []*repository.Asset, constraints []*repository.UniqueConstraint, err error) {
 	events = make([]*repository.Event, len(pushEvents))
 	for i, event := range pushEvents {
 		data, err := EventData(event)
 		if err != nil {
-			return nil, nil, err
+			return nil, nil, nil, err
 		}
 		events[i] = &repository.Event{
 			AggregateID:   event.Aggregate().ID,
@@ -77,9 +77,12 @@ func eventsToRepository(pushEvents []EventPusher) (events []*repository.Event, c
 		if len(event.UniqueConstraints()) > 0 {
 			constraints = append(constraints, uniqueConstraintsToRepository(event.UniqueConstraints())...)
 		}
+		if len(event.Assets()) > 0 {
+			assets = append(assets, assetsToRepository(event.Assets())...)
+		}
 	}
 
-	return events, constraints, nil
+	return events, assets, constraints, nil
 }
 
 func uniqueConstraintsToRepository(constraints []*EventUniqueConstraint) (uniqueConstraints []*repository.UniqueConstraint) {
@@ -95,6 +98,18 @@ func uniqueConstraintsToRepository(constraints []*EventUniqueConstraint) (unique
 	return uniqueConstraints
 }
 
+func assetsToRepository(assets []*Asset) (result []*repository.Asset) {
+	result = make([]*repository.Asset, len(assets))
+	for i, asset := range assets {
+		result[i] = &repository.Asset{
+			ID:     asset.ID,
+			Asset:  asset.Asset,
+			Action: assetActionToRepository(asset.Action),
+		}
+	}
+	return result
+}
+
 //FilterEvents filters the stored events based on the searchQuery
 // and maps the events to the defined event structs
 func (es *Eventstore) FilterEvents(ctx context.Context, queryFactory *SearchQueryBuilder) ([]EventReader, error) {
@@ -229,3 +244,14 @@ func uniqueConstraintActionToRepository(action UniqueConstraintAction) repositor
 		return repository.UniqueConstraintAdd
 	}
 }
+
+func assetActionToRepository(action AssetAction) repository.AssetAction {
+	switch action {
+	case AssetAdd:
+		return repository.AssetAdded
+	case AssetRemove:
+		return repository.AssetRemoved
+	default:
+		return repository.AssetAdded
+	}
+}

internal/eventstore/eventstore_test.go

@@ -68,6 +68,10 @@ func (e *testEvent) UniqueConstraints() []*EventUniqueConstraint {
 	return nil
 }
 
+func (e *testEvent) Assets() []*Asset {
+	return nil
+}
+
 func testFilterMapper(event *repository.Event) (EventReader, error) {
 	if event == nil {
 		return newTestEvent("testID", "hodor", nil, false), nil
@@ -538,7 +542,7 @@ func TestEventstore_aggregatesToEvents(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			events, _, err := eventsToRepository(tt.args.events)
+			events, _, _, err := eventsToRepository(tt.args.events)
 			if (err != nil) != tt.res.wantErr {
 				t.Errorf("Eventstore.aggregatesToEvents() error = %v, wantErr %v", err, tt.res.wantErr)
 				return
@@ -571,7 +575,7 @@ func (repo *testRepo) Health(ctx context.Context) error {
 	return nil
 }
 
-func (repo *testRepo) Push(ctx context.Context, events []*repository.Event, uniqueConstraints ...*repository.UniqueConstraint) error {
+func (repo *testRepo) Push(ctx context.Context, events []*repository.Event, assets []*repository.Asset, uniqueConstraints ...*repository.UniqueConstraint) error {
 	if repo.err != nil {
 		return repo.err
 	}

internal/eventstore/example_test.go

@@ -66,6 +66,10 @@ func (e *UserAddedEvent) UniqueConstraints() []*eventstore.EventUniqueConstraint
 	return nil
 }
 
+func (e *UserAddedEvent) Assets() []*eventstore.Asset {
+	return nil
+}
+
 // ------------------------------------------------------------
 // User first name changed event start
 // ------------------------------------------------------------
@@ -107,6 +111,10 @@ func (e *UserFirstNameChangedEvent) UniqueConstraints() []*eventstore.EventUniqu
 	return nil
 }
 
+func (e *UserFirstNameChangedEvent) Assets() []*eventstore.Asset {
+	return nil
+}
+
 // ------------------------------------------------------------
 // User password checked event start
 // ------------------------------------------------------------
@@ -140,6 +148,10 @@ func (e *UserPasswordCheckedEvent) UniqueConstraints() []*eventstore.EventUnique
 	return nil
 }
 
+func (e *UserPasswordCheckedEvent) Assets() []*eventstore.Asset {
+	return nil
+}
+
 // ------------------------------------------------------------
 // User deleted event
 // ------------------------------------------------------------
@@ -173,6 +185,10 @@ func (e *UserDeletedEvent) UniqueConstraints() []*eventstore.EventUniqueConstrai
 	return nil
 }
 
+func (e *UserDeletedEvent) Assets() []*eventstore.Asset {
+	return nil
+}
+
 // ------------------------------------------------------------
 // Users read model start
 // ------------------------------------------------------------

internal/eventstore/repository/asset.go

@@ -0,0 +1,24 @@
+package repository
+
+//Asset represents all information about a asset (img)
+type Asset struct {
+	// ID is to refer to the asset
+	ID string
+	//Asset is the actual image
+	Asset []byte
+	//Action defines if asset should be added or removed
+	Action AssetAction
+}
+
+type AssetAction int32
+
+const (
+	AssetAdded AssetAction = iota
+	AssetRemoved
+
+	assetCount
+)
+
+func (f AssetAction) Valid() bool {
+	return f >= 0 && f < assetCount
+}

internal/eventstore/repository/mock/repository.mock.go

@@ -79,10 +79,10 @@ func (mr *MockRepositoryMockRecorder) LatestSequence(arg0, arg1 interface{}) *go
 }
 
 // Push mocks base method
-func (m *MockRepository) Push(arg0 context.Context, arg1 []*repository.Event, arg2 ...*repository.UniqueConstraint) error {
+func (m *MockRepository) Push(arg0 context.Context, arg1 []*repository.Event, arg2 []*repository.Asset, arg3 ...*repository.UniqueConstraint) error {
 	m.ctrl.T.Helper()
-	varargs := []interface{}{arg0, arg1}
-	for _, a := range arg2 {
+	varargs := []interface{}{arg0, arg1, arg2}
+	for _, a := range arg3 {
 		varargs = append(varargs, a)
 	}
 	ret := m.ctrl.Call(m, "Push", varargs...)
@@ -91,8 +91,8 @@ func (m *MockRepository) Push(arg0 context.Context, arg1 []*repository.Event, ar
 }
 
 // Push indicates an expected call of Push
-func (mr *MockRepositoryMockRecorder) Push(arg0, arg1 interface{}, arg2 ...interface{}) *gomock.Call {
+func (mr *MockRepositoryMockRecorder) Push(arg0, arg1, arg2 interface{}, arg3 ...interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	varargs := append([]interface{}{arg0, arg1}, arg2...)
+	varargs := append([]interface{}{arg0, arg1, arg2}, arg3...)
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Push", reflect.TypeOf((*MockRepository)(nil).Push), varargs...)
 }

internal/eventstore/repository/mock/repository.mock.impl.go

@@ -24,28 +24,32 @@ func (m *MockRepository) ExpectFilterEvents(events ...*repository.Event) *MockRe
 	return m
 }
 
-func (m *MockRepository) ExpectPush(expectedEvents []*repository.Event, expectedUniqueConstraints ...*repository.UniqueConstraint) *MockRepository {
-	m.EXPECT().Push(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
-		func(ctx context.Context, events []*repository.Event, uniqueConstraints ...*repository.UniqueConstraint) error {
+func (m *MockRepository) ExpectPush(expectedEvents []*repository.Event, expectedAssets []*repository.Asset, expectedUniqueConstraints ...*repository.UniqueConstraint) *MockRepository {
+	m.EXPECT().Push(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
+		func(ctx context.Context, events []*repository.Event, assets []*repository.Asset, uniqueConstraints ...*repository.UniqueConstraint) error {
 			assert.Equal(m.ctrl.T, expectedEvents, events)
 			if expectedUniqueConstraints == nil {
 				expectedUniqueConstraints = []*repository.UniqueConstraint{}
 			}
 			assert.Equal(m.ctrl.T, expectedUniqueConstraints, uniqueConstraints)
+
+			assert.Equal(m.ctrl.T, expectedAssets, assets)
 			return nil
 		},
 	)
 	return m
 }
 
-func (m *MockRepository) ExpectPushFailed(err error, expectedEvents []*repository.Event, expectedUniqueConstraints ...*repository.UniqueConstraint) *MockRepository {
-	m.EXPECT().Push(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
-		func(ctx context.Context, events []*repository.Event, uniqueConstraints ...*repository.UniqueConstraint) error {
+func (m *MockRepository) ExpectPushFailed(err error, expectedEvents []*repository.Event, expectedAssets []*repository.Asset, expectedUniqueConstraints ...*repository.UniqueConstraint) *MockRepository {
+	m.EXPECT().Push(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
+		func(ctx context.Context, events []*repository.Event, assets []*repository.Asset, uniqueConstraints ...*repository.UniqueConstraint) error {
 			assert.Equal(m.ctrl.T, expectedEvents, events)
 			if expectedUniqueConstraints == nil {
 				expectedUniqueConstraints = []*repository.UniqueConstraint{}
 			}
 			assert.Equal(m.ctrl.T, expectedUniqueConstraints, uniqueConstraints)
+
+			assert.Equal(m.ctrl.T, expectedAssets, assets)
 			return err
 		},
 	)

internal/eventstore/repository/repository.go

@@ -9,9 +9,10 @@ type Repository interface {
 	//Health checks if the connection to the storage is available
 	Health(ctx context.Context) error
 	// PushEvents adds all events of the given aggregates to the eventstreams of the aggregates.
+	// if assets are pushed, they will be added to the assets table only referenced by id in eventstore (history)
 	// if unique constraints are pushed, they will be added to the unique table for checking unique constraint violations
 	// This call is transaction save. The transaction will be rolled back if one event fails
-	Push(ctx context.Context, events []*Event, uniqueConstraints ...*UniqueConstraint) error
+	Push(ctx context.Context, events []*Event, assets []*Asset, uniqueConstraints ...*UniqueConstraint) error
 	// Filter returns all events matching the given search query
 	Filter(ctx context.Context, searchQuery *SearchQuery) (events []*Event, err error)
 	//LatestSequence returns the latests sequence found by the the search query

internal/eventstore/repository/sql/crdb.go

@@ -97,9 +97,19 @@ const (
 						$1,
 						$2
 					)`
-
 	uniqueDelete = `DELETE FROM eventstore.unique_constraints
 					WHERE unique_type = $1 and unique_field = $2`
+	assetInsert = `INSERT INTO eventstore.assets
+					(
+						id,
+						asset
+					) 
+					VALUES (  
+						$1,
+						$2
+					)`
+	assetDelete = `DELETE FROM eventstore.assets
+					WHERE id = $1`
 )
 
 type CRDB struct {
@@ -114,7 +124,7 @@ func (db *CRDB) Health(ctx context.Context) error { return db.client.Ping() }
 
 // Push adds all events to the eventstreams of the aggregates.
 // This call is transaction save. The transaction will be rolled back if one event fails
-func (db *CRDB) Push(ctx context.Context, events []*repository.Event, uniqueConstraints ...*repository.UniqueConstraint) error {
+func (db *CRDB) Push(ctx context.Context, events []*repository.Event, assets []*repository.Asset, uniqueConstraints ...*repository.UniqueConstraint) error {
 	err := crdb.ExecuteTx(ctx, db.client, nil, func(tx *sql.Tx) error {
 		stmt, err := tx.PrepareContext(ctx, crdbInsert)
 		if err != nil {
@@ -152,6 +162,11 @@ func (db *CRDB) Push(ctx context.Context, events []*repository.Event, uniqueCons
 		if err != nil {
 			return err
 		}
+
+		err = db.handleAssets(ctx, tx, assets...)
+		if err != nil {
+			return err
+		}
 		return nil
 	})
 	if err != nil && !errors.Is(err, &caos_errs.CaosError{}) {
@@ -194,6 +209,32 @@ func (db *CRDB) handleUniqueConstraints(ctx context.Context, tx *sql.Tx, uniqueC
 	return nil
 }
 
+// handleAssets adds or removes an asset
+func (db *CRDB) handleAssets(ctx context.Context, tx *sql.Tx, assets ...*repository.Asset) (err error) {
+	if assets == nil || len(assets) == 0 || (len(assets) == 1 && assets[0] == nil) {
+		return nil
+	}
+
+	for _, asset := range assets {
+		if asset.Action == repository.AssetAdded {
+			_, err := tx.ExecContext(ctx, assetInsert, asset.ID, asset.Asset)
+			if err != nil {
+				logging.LogWithFields("SQL-M39fs",
+					"asset-id", asset.ID).WithError(err).Info("insert asset failed")
+				return caos_errs.ThrowInternal(err, "SQL-4M0gs", "unable to create asset")
+			}
+		} else if asset.Action == repository.AssetRemoved {
+			_, err := tx.ExecContext(ctx, assetDelete, asset.ID)
+			if err != nil {
+				logging.LogWithFields("SQL-3M9fs",
+					"asset-id", asset.ID).WithError(err).Info("delete asset failed")
+				return caos_errs.ThrowInternal(err, "SQL-Md9ds", "unable to remove unique constraint ")
+			}
+		}
+	}
+	return nil
+}
+
 // Filter returns all events matching the given search query
 func (db *CRDB) Filter(ctx context.Context, searchQuery *repository.SearchQuery) (events []*repository.Event, err error) {
 	events = []*repository.Event{}

internal/eventstore/repository/sql/crdb_test.go

@@ -5,9 +5,10 @@ import (
 	"sync"
 	"testing"
 
-	"github.com/caos/zitadel/internal/eventstore/repository"
 	"github.com/lib/pq"
 	_ "github.com/lib/pq"
+
+	"github.com/caos/zitadel/internal/eventstore/repository"
 )
 
 func TestCRDB_placeholder(t *testing.T) {
@@ -269,12 +270,16 @@ func TestCRDB_Push_OneAggregate(t *testing.T) {
 		ctx               context.Context
 		events            []*repository.Event
 		uniqueConstraints *repository.UniqueConstraint
+		assets            *repository.Asset
 		uniqueDataType    string
 		uniqueDataField   string
+		assetID           string
+		asset             []byte
 	}
 	type eventsRes struct {
 		pushedEventsCount int
 		uniqueCount       int
+		assetCount        int
 		aggType           repository.AggregateType
 		aggID             []string
 	}
@@ -376,6 +381,46 @@ func TestCRDB_Push_OneAggregate(t *testing.T) {
 					aggType:           repository.AggregateType(t.Name()),
 				}},
 		},
+		{
+			name: "push 1 event and add asset",
+			args: args{
+				ctx: context.Background(),
+				events: []*repository.Event{
+					generateEvent(t, "12"),
+				},
+				assets:  generateAddAsset(t, "asset12", []byte{1}),
+				assetID: "asset12",
+				asset:   []byte{1},
+			},
+			res: res{
+				wantErr: false,
+				eventsRes: eventsRes{
+					pushedEventsCount: 1,
+					assetCount:        1,
+					aggID:             []string{"12"},
+					aggType:           repository.AggregateType(t.Name()),
+				}},
+		},
+		{
+			name: "push 1 event and remove asset",
+			args: args{
+				ctx: context.Background(),
+				events: []*repository.Event{
+					generateEvent(t, "13"),
+				},
+				assets:  generateRemoveAsset(t, "asset13"),
+				assetID: "asset13",
+				asset:   []byte{1},
+			},
+			res: res{
+				wantErr: false,
+				eventsRes: eventsRes{
+					pushedEventsCount: 1,
+					assetCount:        0,
+					aggID:             []string{"13"},
+					aggType:           repository.AggregateType(t.Name()),
+				}},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
@@ -389,7 +434,14 @@ func TestCRDB_Push_OneAggregate(t *testing.T) {
 					return
 				}
 			}
-			if err := db.Push(tt.args.ctx, tt.args.events, tt.args.uniqueConstraints); (err != nil) != tt.res.wantErr {
+			if tt.args.uniqueDataType != "" && tt.args.uniqueDataField != "" {
+				err := fillAssets(tt.args.assetID, tt.args.asset)
+				if err != nil {
+					t.Error("unable to prefill insert unique data: ", err)
+					return
+				}
+			}
+			if err := db.Push(tt.args.ctx, tt.args.events, []*repository.Asset{tt.args.assets}, tt.args.uniqueConstraints); (err != nil) != tt.res.wantErr {
 				t.Errorf("CRDB.Push() error = %v, wantErr %v", err, tt.res.wantErr)
 			}
 
@@ -415,7 +467,18 @@ func TestCRDB_Push_OneAggregate(t *testing.T) {
 					t.Errorf("expected unique count %d got %d", tt.res.eventsRes.uniqueCount, uniqueCount)
 				}
 			}
-
+			if tt.args.assets != nil {
+				countAssetRow := testCRDBClient.QueryRow("SELECT COUNT(*) FROM eventstore.assets where id = $1", tt.args.assets.ID)
+				var assetCount int
+				err := countAssetRow.Scan(&assetCount)
+				if err != nil {
+					t.Error("unable to query inserted rows: ", err)
+					return
+				}
+				if assetCount != tt.res.eventsRes.assetCount {
+					t.Errorf("expected asset count %d got %d", tt.res.eventsRes.assetCount, assetCount)
+				}
+			}
 		})
 	}
 }
@@ -423,6 +486,7 @@ func TestCRDB_Push_OneAggregate(t *testing.T) {
 func TestCRDB_Push_MultipleAggregate(t *testing.T) {
 	type args struct {
 		events []*repository.Event
+		assets []*repository.Asset
 	}
 	type eventsRes struct {
 		pushedEventsCount int
@@ -507,7 +571,7 @@ func TestCRDB_Push_MultipleAggregate(t *testing.T) {
 			db := &CRDB{
 				client: testCRDBClient,
 			}
-			if err := db.Push(context.Background(), tt.args.events); (err != nil) != tt.res.wantErr {
+			if err := db.Push(context.Background(), tt.args.events, tt.args.assets); (err != nil) != tt.res.wantErr {
 				t.Errorf("CRDB.Push() error = %v, wantErr %v", err, tt.res.wantErr)
 			}
 
@@ -651,7 +715,7 @@ func TestCRDB_Push_Parallel(t *testing.T) {
 			for _, events := range tt.args.events {
 				wg.Add(1)
 				go func(events []*repository.Event) {
-					err := db.Push(context.Background(), events)
+					err := db.Push(context.Background(), events, nil)
 					if err != nil {
 						errsMu.Lock()
 						errs = append(errs, err)
@@ -698,6 +762,7 @@ func TestCRDB_Filter(t *testing.T) {
 	}
 	type fields struct {
 		existingEvents []*repository.Event
+		assets         []*repository.Asset
 	}
 	type res struct {
 		eventCount int
@@ -763,7 +828,7 @@ func TestCRDB_Filter(t *testing.T) {
 			}
 
 			// setup initial data for query
-			if err := db.Push(context.Background(), tt.fields.existingEvents); err != nil {
+			if err := db.Push(context.Background(), tt.fields.existingEvents, tt.fields.assets); err != nil {
 				t.Errorf("error in setup = %v", err)
 				return
 			}
@@ -786,6 +851,7 @@ func TestCRDB_LatestSequence(t *testing.T) {
 	}
 	type fields struct {
 		existingEvents []*repository.Event
+		existingAssets []*repository.Asset
 	}
 	type res struct {
 		sequence uint64
@@ -849,7 +915,7 @@ func TestCRDB_LatestSequence(t *testing.T) {
 			}
 
 			// setup initial data for query
-			if err := db.Push(context.Background(), tt.fields.existingEvents); err != nil {
+			if err := db.Push(context.Background(), tt.fields.existingEvents, tt.fields.existingAssets); err != nil {
 				t.Errorf("error in setup = %v", err)
 				return
 			}
@@ -869,6 +935,7 @@ func TestCRDB_LatestSequence(t *testing.T) {
 func TestCRDB_Push_ResourceOwner(t *testing.T) {
 	type args struct {
 		events []*repository.Event
+		assets []*repository.Asset
 	}
 	type res struct {
 		resourceOwners []string
@@ -991,7 +1058,7 @@ func TestCRDB_Push_ResourceOwner(t *testing.T) {
 			db := &CRDB{
 				client: testCRDBClient,
 			}
-			if err := db.Push(context.Background(), tt.args.events); err != nil {
+			if err := db.Push(context.Background(), tt.args.events, tt.args.assets); err != nil {
 				t.Errorf("CRDB.Push() error = %v", err)
 			}
 
@@ -1093,3 +1160,22 @@ func generateRemoveUniqueConstraint(t *testing.T, table, uniqueField string) *re
 
 	return e
 }
+
+func generateAddAsset(t *testing.T, id string, asset []byte) *repository.Asset {
+	t.Helper()
+	e := &repository.Asset{
+		ID:     id,
+		Asset:  asset,
+		Action: repository.AssetAdded,
+	}
+	return e
+}
+
+func generateRemoveAsset(t *testing.T, id string) *repository.Asset {
+	t.Helper()
+	e := &repository.Asset{
+		ID:     id,
+		Action: repository.AssetRemoved,
+	}
+	return e
+}

internal/eventstore/repository/sql/local_crdb_test.go

@@ -100,6 +100,11 @@ func fillUniqueData(unique_type, field string) error {
 	return err
 }
 
+func fillAssets(id string, asset []byte) error {
+	_, err := testCRDBClient.Exec("INSERT INTO eventstore.assets (id, asset) VALUES ($1, $2)", id, asset)
+	return err
+}
+
 type migrationPaths []string
 
 type version struct {

internal/eventstore/repository/sql/query_test.go

@@ -304,6 +304,7 @@ func Test_query_events_with_crdb(t *testing.T) {
 	}
 	type fields struct {
 		existingEvents []*repository.Event
+		existingAssets []*repository.Asset
 		client         *sql.DB
 	}
 	type res struct {
@@ -521,7 +522,7 @@ func Test_query_events_with_crdb(t *testing.T) {
 			}
 
 			// setup initial data for query
-			if err := db.Push(context.Background(), tt.fields.existingEvents); err != nil {
+			if err := db.Push(context.Background(), tt.fields.existingEvents, tt.fields.existingAssets); err != nil {
 				t.Errorf("error in setup = %v", err)
 				return
 			}