feature (db): Add on conflict support to db.Create(...) (#1612)

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

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

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
+	}
+}

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)
+	})
 }

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())

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 := &timestamp.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)
+			}
+		})
+	}
+}

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)
 

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;

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

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 = ?`
 )

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 {

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