diff --git a/internal/auth/oidc/repository_auth_method.go b/internal/auth/oidc/repository_auth_method.go index da8dbb569..cadba1dd7 100644 --- a/internal/auth/oidc/repository_auth_method.go +++ b/internal/auth/oidc/repository_auth_method.go @@ -2,11 +2,9 @@ package oidc import ( "context" - "database/sql" "encoding/json" "fmt" "net/url" - "strings" "github.com/hashicorp/boundary/internal/db" "github.com/hashicorp/boundary/internal/errors" @@ -72,148 +70,84 @@ func (r *Repository) upsertAccount(ctx context.Context, am *AuthMethod, IdTokenC if err != nil { return nil, errors.Wrap(ctx, err, op) } - - columns := []string{"public_id", "auth_method_id", "issuer", "subject"} - values := []interface{}{ - sql.Named("1", pubId), - sql.Named("2", am.PublicId), - sql.Named("3", iss), - sql.Named("4", sub), - } - var conflictClauses, fieldMasks, nullMasks []string - - { - marshaledTokenClaims, err := json.Marshal(IdTokenClaims) - if err != nil { - return nil, errors.Wrap(ctx, err, op) - } - columns, values = append(columns, "token_claims"), append(values, sql.Named(fmt.Sprintf("%d", len(values)+1), string(marshaledTokenClaims))) - conflictClauses = append(conflictClauses, fmt.Sprintf("token_claims = @%d", len(values))) - fieldMasks = append(fieldMasks, TokenClaimsField) - } - { - marshaledAccessTokenClaims, err := json.Marshal(AccessTokenClaims) - if err != nil { - return nil, errors.Wrap(ctx, err, op) - } - columns, values = append(columns, "userinfo_claims"), append(values, sql.Named(fmt.Sprintf("%d", len(values)+1), string(marshaledAccessTokenClaims))) - conflictClauses = append(conflictClauses, fmt.Sprintf("userinfo_claims = @%d", len(values))) - fieldMasks = append(fieldMasks, UserinfoClaimsField) - } - issAsUrl, err := url.Parse(iss) if err != nil { return nil, errors.New(ctx, errors.Unknown, op, "unable to parse issuer", errors.WithWrap(err)) } - acctForOplog, err := NewAccount(ctx, am.PublicId, sub, WithIssuer(issAsUrl)) - if err != nil { - return nil, errors.Wrap(ctx, err, op, errors.WithMsg("unable to create new acct for oplog")) + + acct := AllocAccount() + acct.PublicId = pubId + acct.AuthMethodId = am.PublicId + acct.Issuer = issAsUrl.String() + acct.Subject = sub + + oc := db.OnConflict{ + Target: db.Constraint("auth_oidc_account_auth_method_id_issuer_subject_uq"), + Action: db.SetColumns([]string{"public_id", "auth_method_id", "issuer", "subject", "userinfo_claims", "token_claims"}), } - var foundName interface{} + marshaledTokenClaims, err := json.Marshal(IdTokenClaims) + if err != nil { + return nil, errors.Wrap(ctx, err, op) + } + acct.TokenClaims = string(marshaledTokenClaims) + + marshaledAccessTokenClaims, err := json.Marshal(AccessTokenClaims) + if err != nil { + return nil, errors.Wrap(ctx, err, op) + } + acct.UserinfoClaims = string(marshaledAccessTokenClaims) + + var foundName, foundEmail bool switch { case AccessTokenClaims[fromName] != nil: - foundName = AccessTokenClaims[fromName] - columns, values = append(columns, "full_name"), append(values, sql.Named(fmt.Sprintf("%d", len(values)+1), foundName)) + foundName = true + acct.FullName = AccessTokenClaims[fromName].(string) case IdTokenClaims[fromName] != nil: - foundName = IdTokenClaims[fromName] - columns, values = append(columns, "full_name"), append(values, sql.Named(fmt.Sprintf("%d", len(values)+1), foundName)) - } - if foundName != nil { - acctForOplog.FullName = foundName.(string) - conflictClauses = append(conflictClauses, fmt.Sprintf("full_name = @%d", len(values))) - fieldMasks = append(fieldMasks, NameField) - } else { - conflictClauses = append(conflictClauses, "full_name = NULL") - nullMasks = append(nullMasks, NameField) + foundName = true + acct.FullName = IdTokenClaims[fromName].(string) } - var foundEmail interface{} switch { case AccessTokenClaims[fromEmail] != nil: - foundEmail = AccessTokenClaims[fromEmail] - columns, values = append(columns, "email"), append(values, sql.Named(fmt.Sprintf("%d", len(values)+1), foundEmail)) + foundEmail = true + acct.Email = AccessTokenClaims[fromEmail].(string) case IdTokenClaims[fromEmail] != nil: - foundEmail = IdTokenClaims[fromEmail] - columns, values = append(columns, "email"), append(values, sql.Named(fmt.Sprintf("%d", len(values)+1), foundEmail)) + foundEmail = true + acct.Email = IdTokenClaims[fromEmail].(string) } - if foundEmail != nil { - acctForOplog.Email = foundEmail.(string) - conflictClauses = append(conflictClauses, fmt.Sprintf("email = @%d", len(values))) - fieldMasks = append(fieldMasks, "Email") - } else { - conflictClauses = append(conflictClauses, "email = NULL") - nullMasks = append(nullMasks, "Email") + if foundName { + oc.Action = append(oc.Action.([]db.ColumnValue), db.SetColumns([]string{"full_name"})...) } - - placeHolders := make([]string, 0, len(columns)) - for colNum := range columns { - placeHolders = append(placeHolders, fmt.Sprintf("@%d", colNum+1)) + if foundEmail { + oc.Action = append(oc.Action.([]db.ColumnValue), db.SetColumns([]string{"email"})...) } - query := fmt.Sprintf(acctUpsertQuery, strings.Join(columns, ", "), strings.Join(placeHolders, ", "), strings.Join(conflictClauses, ", ")) oplogWrapper, err := r.kms.GetWrapper(ctx, am.ScopeId, kms.KeyPurposeOplog) if err != nil { return nil, errors.Wrap(ctx, err, op, errors.WithMsg("unable to get oplog wrapper")) } - updatedAcct := AllocAccount() _, err = r.writer.DoTx( ctx, db.StdRetryCnt, db.ExpBackoff{}, func(reader db.Reader, w db.Writer) error { - var err error - rows, err := w.Query(ctx, query, values) + var rowCnt int64 + err := w.Create(ctx, acct, db.WithOnConflict(&oc), db.WithReturnRowsAffected(&rowCnt), db.WithOplog(oplogWrapper, acct.oplog(oplog.OpType_OP_TYPE_CREATE, am.ScopeId))) if err != nil { return errors.Wrap(ctx, err, op, errors.WithMsg("unable to insert/update auth oidc account")) } - defer rows.Close() - result := struct { - PublicId string - Version int - }{} - var rowCnt int - for rows.Next() { - rowCnt += 1 - err = r.reader.ScanRows(rows, &result) - if err != nil { - return errors.Wrap(ctx, err, op, errors.WithMsg("unable to scan rows for account")) - } - } if rowCnt > 1 { return errors.New(ctx, errors.MultipleRecords, op, fmt.Sprintf("expected 1 row but got: %d", rowCnt)) } - if err := reader.LookupWhere(ctx, &updatedAcct, "auth_method_id = ? and issuer = ? and subject = ?", am.PublicId, iss, sub); err != nil { - return errors.Wrap(ctx, err, op, errors.WithMsg(fmt.Sprintf("unable to look up auth oidc account for: %s / %s / %s", am.PublicId, iss, sub))) - } - // include the version incase of predictable account public ids based on a calculation using authmethod id and subject - if result.Version == 1 && updatedAcct.PublicId == pubId { - if err := upsertOplog(ctx, w, oplogWrapper, oplog.OpType_OP_TYPE_CREATE, am.ScopeId, updatedAcct, nil, nil); err != nil { - return errors.Wrap(ctx, err, op, errors.WithMsg("unable to write create oplog for account")) - } - } else { - if len(fieldMasks) > 0 || len(nullMasks) > 0 { - acctForOplog := AllocAccount() - acctForOplog.PublicId = updatedAcct.PublicId - if foundEmail != nil { - acctForOplog.Email = foundEmail.(string) - } - if foundName != nil { - acctForOplog.FullName = foundName.(string) - } - if err := upsertOplog(ctx, w, oplogWrapper, oplog.OpType_OP_TYPE_UPDATE, am.ScopeId, acctForOplog, fieldMasks, nullMasks); err != nil { - return errors.Wrap(ctx, err, op, errors.WithMsg("unable to write update oplog for account")) - } - } - } return nil }, ) if err != nil { return nil, errors.Wrap(ctx, err, op) } - return updatedAcct, nil + return acct, nil } // upsertOplog will write oplog msgs for account upserts. The db.Writer needs to be the writer for the current diff --git a/internal/db/clause.go b/internal/db/clause.go new file mode 100644 index 000000000..0beb4d758 --- /dev/null +++ b/internal/db/clause.go @@ -0,0 +1,123 @@ +package db + +import ( + "sort" + + "gorm.io/gorm" + "gorm.io/gorm/clause" +) + +// ColumnValue defines a column and it's assigned value for a database operation +type ColumnValue struct { + column string + value interface{} +} + +type assigner interface { + toAssignment(column string) clause.Assignment +} + +// column represents a table column +type column struct { + name string + table string +} + +func (c *column) toAssignment(column string) clause.Assignment { + return clause.Assignment{ + Column: clause.Column{Name: column}, + Value: clause.Column{Table: c.table, Name: c.name}, + } +} + +func rawAssignment(column string, value interface{}) clause.Assignment { + return clause.Assignment{ + Column: clause.Column{Name: column}, + Value: value, + } +} + +// ExprValue encapsulates an expression value for a column assignment. See +// Expr(...) to create these values. +type ExprValue struct { + sql string + vars []interface{} +} + +func (ev *ExprValue) toAssignment(column string) clause.Assignment { + return clause.Assignment{ + Column: clause.Column{Name: column}, + Value: gorm.Expr(ev.sql, ev.vars...), + } +} + +// Expr creates an expression value (ExprValue) which can be used when setting +// column values for database operations. See: Expr(...) +// +// Set name column to null example: +// SetColumnValues(map[string]interface{}{"name": Expr("NULL")}) +// +// Set exp_time column to N seconds from now: +// SetColumnValues(map[string]interface{}{"exp_time": Expr("wt_add_seconds_to_now(?)", 10)}) +func Expr(expr string, args ...interface{}) ExprValue { + return ExprValue{sql: expr, vars: args} +} + +// SetColumnValues defines a map from column names to values +func SetColumnValues(columnValues map[string]interface{}) []ColumnValue { + keys := make([]string, 0, len(columnValues)) + for key := range columnValues { + keys = append(keys, key) + } + sort.Strings(keys) + + assignments := make([]ColumnValue, len(keys)) + for idx, key := range keys { + assignments[idx] = ColumnValue{column: key, value: columnValues[key]} + } + return assignments +} + +// SetColumns defines a list of column (names) to update using the set of +// proposed insert columns during an on conflict update. +func SetColumns(names []string) []ColumnValue { + assignments := make([]ColumnValue, len(names)) + for idx, name := range names { + assignments[idx] = ColumnValue{ + column: name, + value: column{name: name, table: "excluded"}, + } + } + return assignments +} + +// OnConflict specifies how to handle alternative actions to take when an insert +// results in a unique constraint or exclusion constraint error. +type OnConflict struct { + + // Target specifies what conflict you want to define a policy for. This can + // be any one of these: + // Columns: the name of a specific column or columns + // Constraint: the name of a unique constraint + Target interface{} + + // Action specifies the action to take on conflict. This can be any one of + // these: + // DoNothing: leaves the conflicting record as-is + // UpdateAll: updates all the columns of the conflicting record using the resource's data + // []ColumnValue: update a set of columns of the conflicting record using the set of assignments + Action interface{} +} + +// Constraint defines database constraint name +type Constraint string + +// Columns defines a set of column names +type Columns []string + +// DoNothing defines an "on conflict" action of doing nothing +type DoNothing bool + +// UpdateAll defines an "on conflict" action of updating all columns using the +// proposed insert column values +type UpdateAll bool diff --git a/internal/db/option.go b/internal/db/option.go index 32b50dcc4..ba963875d 100644 --- a/internal/db/option.go +++ b/internal/db/option.go @@ -47,6 +47,9 @@ type Options struct { withGormFormatter hclog.Logger withMaxOpenConnections int + + withOnConflict *OnConflict + withRowsAffected *int64 } type oplogOpts struct { @@ -185,3 +188,19 @@ func WithMaxOpenConnections(max int) Option { o.withMaxOpenConnections = max } } + +// WithOnConflict specifies an optional on conflict criteria which specify +// alternative actions to take when an insert results in a unique constraint or +// exclusion constraint error +func WithOnConflict(onConflict *OnConflict) Option { + return func(o *Options) { + o.withOnConflict = onConflict + } +} + +// WithReturnRowsAffected specifies an option for returning the rows affected +func WithReturnRowsAffected(rowsAffected *int64) Option { + return func(o *Options) { + o.withRowsAffected = rowsAffected + } +} diff --git a/internal/db/option_test.go b/internal/db/option_test.go index 963b9d008..f2cf9588b 100644 --- a/internal/db/option_test.go +++ b/internal/db/option_test.go @@ -141,8 +141,6 @@ func Test_getOpts(t *testing.T) { testOpts := getDefaultOptions() testOpts.withSkipVetForWrite = false assert.Equal(opts, testOpts) - - // try setting to false opts = GetOpts(WithSkipVetForWrite(true)) testOpts = getDefaultOptions() testOpts.withSkipVetForWrite = true @@ -156,8 +154,6 @@ func Test_getOpts(t *testing.T) { testOpts.withWhereClause = "" testOpts.withWhereClauseArgs = nil assert.Equal(opts, testOpts) - - // try setting to false opts = GetOpts(WithWhere("id = ? and foo = ?", 1234, "bar")) testOpts.withWhereClause = "id = ? and foo = ?" testOpts.withWhereClauseArgs = []interface{}{1234, "bar"} @@ -170,8 +166,6 @@ func Test_getOpts(t *testing.T) { testOpts := getDefaultOptions() testOpts.withOrder = "" assert.Equal(opts, testOpts) - - // try setting to false opts = GetOpts(WithOrder("version desc")) testOpts.withOrder = "version desc" assert.Equal(opts, testOpts) @@ -184,8 +178,6 @@ func Test_getOpts(t *testing.T) { assert.Equal(opts, testOpts) testLogger := hclog.New(&hclog.LoggerOptions{}) - - // try setting to false opts = GetOpts(WithGormFormatter(testLogger)) testOpts.withGormFormatter = testLogger assert.Equal(opts, testOpts) @@ -196,10 +188,37 @@ func Test_getOpts(t *testing.T) { opts := GetOpts() testOpts := getDefaultOptions() assert.Equal(opts, testOpts) - - // try setting to false opts = GetOpts(WithMaxOpenConnections(22)) testOpts.withMaxOpenConnections = 22 assert.Equal(opts, testOpts) }) + t.Run("WithOnConflict", func(t *testing.T) { + assert := assert.New(t) + // test default of false + opts := GetOpts() + testOpts := getDefaultOptions() + assert.Equal(opts, testOpts) + + columns := SetColumns([]string{"name", "description"}) + columnValues := SetColumnValues(map[string]interface{}{"expiration": "NULL"}) + testOnConflict := OnConflict{ + Target: Constraint("uniq-name"), + Action: append(columns, columnValues...), + } + opts = GetOpts(WithOnConflict(&testOnConflict)) + testOpts.withOnConflict = &testOnConflict + assert.Equal(opts, testOpts) + }) + t.Run("WithReturnRowsAffected", func(t *testing.T) { + assert := assert.New(t) + // test default of false + opts := GetOpts() + testOpts := getDefaultOptions() + assert.Equal(opts, testOpts) + + var rowsAffected int64 + opts = GetOpts(WithReturnRowsAffected(&rowsAffected)) + testOpts.withRowsAffected = &rowsAffected + assert.Equal(opts, testOpts) + }) } diff --git a/internal/db/read_writer.go b/internal/db/read_writer.go index 119bb8f42..e942147ae 100644 --- a/internal/db/read_writer.go +++ b/internal/db/read_writer.go @@ -15,6 +15,7 @@ import ( wrapping "github.com/hashicorp/go-kms-wrapping" "google.golang.org/protobuf/proto" "gorm.io/gorm" + "gorm.io/gorm/clause" ) const ( @@ -260,10 +261,12 @@ func (rw *Db) lookupAfterWrite(ctx context.Context, i interface{}, opt ...Option return nil } -// Create an object in the db with options: WithOplog, NewOplogMsg and -// WithLookup. WithOplog will write an oplog entry for the create. +// Create an object in the db with options: WithOplog, NewOplogMsg, WithLookup +// and OnConflict. WithOplog will write an oplog entry for the create. // NewOplogMsg will return in-memory oplog message. WithOplog and NewOplogMsg // cannot be used together. WithLookup with to force a lookup after create. +// OnConflict specifies alternative actions to take when an insert results in a +// unique constraint or exclusion constraint error. func (rw *Db) Create(ctx context.Context, i interface{}, opt ...Option) error { const op = "db.Create" if rw.underlying == nil { @@ -295,6 +298,54 @@ func (rw *Db) Create(ctx context.Context, i interface{}, opt ...Option) error { } } } + + db := rw.underlying.WithContext(ctx) + var onConflictDoNothing bool + if opts.withOnConflict != nil { + c := clause.OnConflict{} + switch opts.withOnConflict.Target.(type) { + case Constraint: + c.OnConstraint = string(opts.withOnConflict.Target.(Constraint)) + case Columns: + columns := make([]clause.Column, 0, len(opts.withOnConflict.Target.(Columns))) + for _, name := range opts.withOnConflict.Target.(Columns) { + columns = append(columns, clause.Column{Name: name}) + } + c.Columns = columns + default: + return errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("invalid conflict target: %v", reflect.TypeOf(opts.withOnConflict.Target))) + } + + switch opts.withOnConflict.Action.(type) { + case DoNothing: + c.DoNothing = true + onConflictDoNothing = true + case UpdateAll: + c.UpdateAll = true + case []ColumnValue: + updates := opts.withOnConflict.Action.([]ColumnValue) + set := make(clause.Set, 0, len(updates)) + for _, s := range updates { + // make sure it's not one of the std immutable columns + if contains([]string{"createtime", "publicid"}, strings.ToLower(s.column)) { + return errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("Cannot do update on conflict for column %s", s.column)) + } + switch sv := s.value.(type) { + case column: + set = append(set, sv.toAssignment(s.column)) + case ExprValue: + set = append(set, sv.toAssignment(s.column)) + default: + set = append(set, rawAssignment(s.column, s.value)) + } + } + c.DoUpdates = set + default: + return errors.New(ctx, errors.InvalidParameter, op, fmt.Sprintf("invalid conflict action: %v", reflect.TypeOf(opts.withOnConflict.Action))) + } + db = db.Clauses(c) + } + var ticket *store.Ticket if withOplog { var err error @@ -303,20 +354,32 @@ func (rw *Db) Create(ctx context.Context, i interface{}, opt ...Option) error { return errors.Wrap(ctx, err, op, errors.WithMsg("unable to get ticket")) } } - if err := rw.underlying.Create(i).Error; err != nil { - return errors.Wrap(ctx, err, op, errors.WithMsg("create failed"), errors.WithoutEvent()) + tx := db.Create(i) + if tx.Error != nil { + return errors.Wrap(ctx, tx.Error, op, errors.WithMsg("create failed"), errors.WithoutEvent()) + } + if opts.withRowsAffected != nil { + *opts.withRowsAffected = tx.RowsAffected } if withOplog { - if err := rw.addOplog(ctx, CreateOp, opts, ticket, i); err != nil { - return errors.Wrap(ctx, err, op) + switch { + case onConflictDoNothing && tx.RowsAffected == 0: + default: + if err := rw.addOplog(ctx, CreateOp, opts, ticket, i); err != nil { + return errors.Wrap(ctx, err, op) + } } } if opts.newOplogMsg != nil { - msg, err := rw.newOplogMessage(ctx, CreateOp, i) - if err != nil { - return errors.Wrap(ctx, err, op, errors.WithMsg("returning oplog failed")) + switch { + case onConflictDoNothing && tx.RowsAffected == 0: + default: + msg, err := rw.newOplogMessage(ctx, CreateOp, i) + if err != nil { + return errors.Wrap(ctx, err, op, errors.WithMsg("returning oplog failed")) + } + *opts.newOplogMsg = *msg } - *opts.newOplogMsg = *msg } if err := rw.lookupAfterWrite(ctx, i, opt...); err != nil { return errors.Wrap(ctx, err, op, errors.WithoutEvent()) diff --git a/internal/db/read_writer_ext_test.go b/internal/db/read_writer_ext_test.go new file mode 100644 index 000000000..889ecdaaa --- /dev/null +++ b/internal/db/read_writer_ext_test.go @@ -0,0 +1,152 @@ +package db_test + +import ( + "context" + "testing" + "time" + + "github.com/hashicorp/boundary/internal/db" + "github.com/hashicorp/boundary/internal/db/db_test" + "github.com/hashicorp/boundary/internal/db/timestamp" + "github.com/hashicorp/boundary/internal/oplog" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/types/known/timestamppb" +) + +func TestDb_Create_OnConflict(t *testing.T) { + ctx := context.Background() + wrapper := db.TestWrapper(t) + conn, _ := db.TestSetup(t, "postgres") + rw := db.New(conn) + + // create initial user for on conflict tests + id, err := db.NewPublicId("test-user") + require.NoError(t, err) + initialUser, err := db_test.NewTestUser() + require.NoError(t, err) + ts := ×tamp.Timestamp{Timestamp: timestamppb.Now()} + initialUser.CreateTime = ts + initialUser.UpdateTime = ts + initialUser.Name = "foo-" + id + err = rw.Create(ctx, initialUser) + require.NoError(t, err) + assert.NotEmpty(t, initialUser.Id) + + tests := []struct { + name string + onConflict db.OnConflict + wantUpdate bool + wantEmail string + wantOplog bool + }{ + { + name: "set-columns", + onConflict: db.OnConflict{ + Target: db.Columns{"public_id"}, + Action: db.SetColumns([]string{"name"}), + }, + wantUpdate: true, + wantOplog: true, + }, + { + name: "set-column-values", + onConflict: db.OnConflict{ + Target: db.Columns{"public_id"}, + Action: db.SetColumnValues(map[string]interface{}{ + "name": db.Expr("md5(?)", "alice eve smith"), + "email": "alice@gmail.com", + "phone_number": db.Expr("NULL"), + }), + }, + wantUpdate: true, + wantEmail: "alice@gmail.com", + wantOplog: true, + }, + { + name: "both-set-columns-and-set-column-values", + onConflict: func() db.OnConflict { + onConflict := db.OnConflict{ + Target: db.Columns{"public_id"}, + } + cv := db.SetColumns([]string{"name"}) + cv = append(cv, + db.SetColumnValues(map[string]interface{}{ + "email": "alice@gmail.com", + "phone_number": db.Expr("NULL"), + })...) + onConflict.Action = cv + return onConflict + }(), + wantUpdate: true, + wantEmail: "alice@gmail.com", + wantOplog: true, + }, + { + name: "update-all", + onConflict: db.OnConflict{ + Target: db.Columns{"public_id"}, + Action: db.UpdateAll(true), + }, + wantUpdate: true, + wantOplog: true, + }, + { + name: "do-nothing", + onConflict: db.OnConflict{ + Target: db.Columns{"public_id"}, + Action: db.DoNothing(true), + }, + wantOplog: false, + }, + { + name: "on-constraint", + onConflict: db.OnConflict{ + Target: db.Constraint("db_test_user_public_id_key"), + Action: db.SetColumns([]string{"name"}), + }, + wantUpdate: true, + wantOplog: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert, require := assert.New(t), require.New(t) + conflictUser, err := db_test.NewTestUser() + require.NoError(err) + userNameId, err := db.NewPublicId("test-user-name") + require.NoError(err) + conflictUser.PublicId = initialUser.PublicId + conflictUser.Name = userNameId + md := oplog.Metadata{ + "resource-public-id": []string{conflictUser.PublicId}, + "op-type": []string{oplog.OpType_OP_TYPE_CREATE.String()}, + } + var rowsAffected int64 + err = rw.Create(ctx, conflictUser, db.WithOnConflict(&tt.onConflict), db.WithOplog(wrapper, md), db.WithReturnRowsAffected(&rowsAffected)) + require.NoError(err) + if tt.wantOplog { + err = db.TestVerifyOplog(t, rw, conflictUser.PublicId, db.WithOperation(oplog.OpType_OP_TYPE_CREATE), db.WithCreateNotBefore(10*time.Second)) + assert.NoError(err) + } + foundUser, err := db_test.NewTestUser() + require.NoError(err) + foundUser.PublicId = conflictUser.PublicId + err = rw.LookupByPublicId(context.Background(), foundUser) + require.NoError(err) + if tt.wantUpdate { + assert.Equal(int64(1), rowsAffected) + assert.Equal(conflictUser.Id, foundUser.Id) + assert.Equal(conflictUser.Name, foundUser.Name) + if tt.wantEmail != "" { + assert.Equal(tt.wantEmail, foundUser.Email) + } + } else { + assert.Equal(int64(0), rowsAffected) + assert.NotEqual(conflictUser.Id, foundUser.Id) + assert.NotEqual(conflictUser.Name, foundUser.Name) + } + }) + } +} diff --git a/internal/db/read_writer_test.go b/internal/db/read_writer_test.go index 14404793c..793663dd4 100644 --- a/internal/db/read_writer_test.go +++ b/internal/db/read_writer_test.go @@ -723,8 +723,8 @@ func TestDb_Create(t *testing.T) { metadata := oplog.Metadata{ "resource-public-id": []string{user.PublicId}, - // "op-type": []string{oplog.OpType_OP_TYPE_UPDATE.String()}, } + err = w.WriteOplogEntryWith(context.Background(), TestWrapper(t), ticket, metadata, []*oplog.Message{&createMsg, &updateMsg}) require.NoError(err) diff --git a/internal/db/schema/migrations/oss/postgres/17/01_servers.up.sql b/internal/db/schema/migrations/oss/postgres/17/01_servers.up.sql new file mode 100644 index 000000000..4b84e5c9c --- /dev/null +++ b/internal/db/schema/migrations/oss/postgres/17/01_servers.up.sql @@ -0,0 +1,16 @@ +begin; + +-- server table inserts/updates historically included setting the update_time +-- column to now(). Since we're now using the "standard" db package to do this, +-- we need to add triggers to set the update_time for every insert/update. +create trigger + update_time_column +before update on server + for each row execute procedure update_time_column(); + +create trigger + insert_time_column +before insert on server + for each row execute procedure update_time_column(); + +commit; \ No newline at end of file diff --git a/internal/proto/local/controller/servers/v1/servers.proto b/internal/proto/local/controller/servers/v1/servers.proto index 7adf4bbb3..ecb426497 100644 --- a/internal/proto/local/controller/servers/v1/servers.proto +++ b/internal/proto/local/controller/servers/v1/servers.proto @@ -17,6 +17,7 @@ message Server { // private ID and name to be identical, and there is currently no reason we // can think of to allow a given server's name to match another within the // same cluster. So we simply use the private ID and deprecate this. + // @inject_tag: `gorm:"-"` string name = 30 [deprecated = true]; // Description of the resource diff --git a/internal/servers/query.go b/internal/servers/query.go index 1e814a853..c0f865a17 100644 --- a/internal/servers/query.go +++ b/internal/servers/query.go @@ -1,18 +1,6 @@ package servers const ( - serverUpsertQuery = ` - insert into server - (private_id, type, description, address, update_time) - values - (@private_id, @type, @description, @address, now()) - on conflict on constraint server_pkey - do update set - type = @type, - description = @description, - address = @address, - update_time = now(); - ` deleteWhereCreateTimeSql = `create_time < ?` deleteTagsSql = `server_id = ?` ) diff --git a/internal/servers/repository.go b/internal/servers/repository.go index 2634e87a7..1a5a49639 100644 --- a/internal/servers/repository.go +++ b/internal/servers/repository.go @@ -2,7 +2,6 @@ package servers import ( "context" - "database/sql" "fmt" "time" @@ -138,7 +137,7 @@ func (r *Repository) UpsertServer(ctx context.Context, server *Server, opt ...Op opts := getOpts(opt...) - var rowsUpdated int + var rowsUpdated int64 var controllers []*Server _, err := r.writer.DoTx( ctx, @@ -146,14 +145,11 @@ func (r *Repository) UpsertServer(ctx context.Context, server *Server, opt ...Op db.ExpBackoff{}, func(read db.Reader, w db.Writer) error { var err error - rowsUpdated, err = w.Exec(ctx, - serverUpsertQuery, - []interface{}{ - sql.Named("private_id", server.PrivateId), - sql.Named("type", server.Type), - sql.Named("description", server.Description), - sql.Named("address", server.Address), - }) + onConflict := &db.OnConflict{ + Target: db.Constraint("server_pkey"), + Action: append(db.SetColumns([]string{"type", "description", "address"}), db.SetColumnValues(map[string]interface{}{"update_time": "now()"})...), + } + err = w.Create(ctx, server, db.WithOnConflict(onConflict), db.WithReturnRowsAffected(&rowsUpdated)) if err != nil { return errors.Wrap(ctx, err, op+":Upsert") } @@ -208,7 +204,7 @@ func (r *Repository) UpsertServer(ctx context.Context, server *Server, opt ...Op return nil, db.NoRowsAffected, err } - return controllers, rowsUpdated, nil + return controllers, int(rowsUpdated), nil } type RecoveryNonce struct { diff --git a/internal/servers/servers.pb.go b/internal/servers/servers.pb.go index b5fad0c29..cd07d8563 100644 --- a/internal/servers/servers.pb.go +++ b/internal/servers/servers.pb.go @@ -35,9 +35,10 @@ type Server struct { // private ID and name to be identical, and there is currently no reason we // can think of to allow a given server's name to match another within the // same cluster. So we simply use the private ID and deprecate this. + // @inject_tag: `gorm:"-"` // // Deprecated: Do not use. - Name string `protobuf:"bytes,30,opt,name=name,proto3" json:"name,omitempty"` + Name string `protobuf:"bytes,30,opt,name=name,proto3" json:"name,omitempty" gorm:"-"` // Description of the resource Description string `protobuf:"bytes,40,opt,name=description,proto3" json:"description,omitempty"` // Address for the server