basic internal db interfaces and Gorm implementation

* db reader and writer interfaces which should allow us to encapsulate the vast majority of database code in this package.

* GormReadWriter implementation of the Reader and Writer interfaces

Makefile

@@ -42,6 +42,7 @@ protobuild:
 	@protoc --proto_path=internal/proto/local --proto_path=internal/proto/third_party --swagger_out=logtostderr=true,disable_default_errors=true,include_package_in_tags=true,fqn_for_swagger_name=true,allow_merge,merge_file_name=controller:internal/gen/. internal/proto/local/controller/api/services/v1/*.proto
 	@protoc-go-inject-tag -input=./internal/oplog/store/oplog.pb.go
 	@protoc-go-inject-tag -input=./internal/oplog/oplog_test/oplog_test.pb.go
+	@protoc-go-inject-tag -input=./internal/db/db_test/db_test.pb.go
 
 
 cleanup:

internal/db/README.md

@@ -0,0 +1,67 @@
+# db package
+
+## Usage
+Just some high-level usage highlights to get you started.  Read the godocs for a complete list of capabilities and their documentation.
+
+```go
+    conn, _ := gorm.Open("postgres", url)
+    
+    // GormReadWriter implements both the Reader and Writer interfaces
+    rw := GormReadWriter{Tx: conn}
+    
+    // There are writer methods like: Create, Update and Delete
+    // that will write Gorm struct to the db.  These writer methods
+    // all support options for writing Oplog entries for the 
+    // caller: WithOplog(yourWrapper, yourMetadata)
+    // the caller is responsible for the transaction life cycle of the writer
+    // and if an error is returned the caller must decide what to do with 
+    // the transaction, which is almost always a rollback for the caller.
+    err = rw.Create(context.Background(), user)
+   
+    // There are reader methods like: LookupByPublicId,  
+    // LookupByFriendlyName, SearchBy, LookupBy, etc
+    // which will lookup resources for you and scan them into your Gorm struct
+    err = rw.LookupByPublicId(context.Background(), foundUser)
+
+    // There's reader ScanRows that facilitates scanning rows from 
+    // a "raw" SQL query into your Gorm struct
+    tx, err := rw.DB()
+    where := "select * from test_users where friendly_name in ($1, $2)"
+    rows, err := tx.Query(where, "alice", "bob")
+	defer rows.Close()
+	for rows.Next() {
+        user := db_test.NewTestUser()
+        // scan the row into your Gorm struct
+		if err := rw.ScanRows(rows, &user); err != nil {
+            return err
+        }
+        // Do something with the Gorm user struct
+    }
+
+    // DoTx is a writer function that wraps a TxHandler 
+    // in a retryable transaction.  You simply implement a
+    // TxHandler that does your writes and hand the handler
+    // to DoTx, which will wrap the writes in a retryable 
+    // transaction with the retry attempts and backoff
+    // strategy you specify via options.
+    _, err = w.DoTx(
+        context.Background(),
+        10,           // ten retries
+        ExpBackoff{}, // exponential backoff
+        func(w Writer) error {
+            // the TxHandler updates the user's friendly name
+            return w.Update(context.Background(), user, []string{"FriendlyName"},
+                // write oplogs for this update
+                WithOplog(
+                    InitTestWrapper(t),
+                    oplog.Metadata{
+                        "key-only":   nil,
+                        "deployment": []string{"amex"},
+                        "project":    []string{"central-info-systems", "local-info-systems"},
+                    },
+                ),
+            )
+        })
+
+
+```

internal/db/backoff.go

@@ -0,0 +1,26 @@
+package db
+
+import (
+	"math"
+	"math/rand"
+	"time"
+)
+
+type Backoff interface {
+	Duration(attemptNumber uint) time.Duration
+}
+
+type ConstBackoff struct {
+	DurationMs time.Duration
+}
+
+func (b ConstBackoff) Duration(attempt uint) time.Duration {
+	return time.Millisecond * time.Duration(b.DurationMs)
+}
+
+type ExpBackoff struct{}
+
+func (b ExpBackoff) Duration(attempt uint) time.Duration {
+	r := rand.Float64()
+	return time.Millisecond * time.Duration(math.Exp2(float64(attempt))*5*(r+0.5))
+}

internal/db/db_test/db.go

@@ -0,0 +1,132 @@
+// Package db_test provides some helper funcs for testing db integrations
+package db_test
+
+import (
+	"database/sql/driver"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/golang/protobuf/ptypes"
+	"github.com/hashicorp/vault/sdk/helper/base62"
+	"google.golang.org/protobuf/proto"
+)
+
+type TestUser struct {
+	*StoreTestUser
+	table string `gorm:"-"`
+}
+
+func NewTestUser() (*TestUser, error) {
+	publicId, err := base62.Random(20)
+	if err != nil {
+		return nil, err
+	}
+	return &TestUser{
+		StoreTestUser: &StoreTestUser{
+			PublicId: publicId,
+		},
+	}, nil
+}
+
+// Clone is useful when you're retrying transactions and you need to send the user several times
+func (u *TestUser) Clone() *TestUser {
+	s := proto.Clone(u.StoreTestUser)
+	return &TestUser{
+		StoreTestUser: s.(*StoreTestUser),
+	}
+}
+func (u *TestUser) TableName() string {
+	if u.table != "" {
+		return u.table
+	}
+	return "db_test_user"
+}
+
+func (u *TestUser) SetTableName(name string) {
+	if name != "" {
+		u.table = name
+	}
+}
+
+type TestCar struct {
+	*StoreTestCar
+	table string `gorm:"-"`
+}
+
+func NewTestCar() (*TestCar, error) {
+	publicId, err := base62.Random(20)
+	if err != nil {
+		return nil, err
+	}
+	return &TestCar{
+		StoreTestCar: &StoreTestCar{
+			PublicId: publicId,
+		},
+	}, nil
+}
+
+func (c *TestCar) TableName() string {
+	if c.table != "" {
+		return c.table
+	}
+
+	return "db_test_car"
+}
+func (c *TestCar) SetTableName(name string) {
+	if name != "" {
+		c.table = name
+	}
+}
+
+type TestRental struct {
+	*StoreTestRental
+	table string `gorm:"-"`
+}
+
+func NewTestRental() (*TestRental, error) {
+	publicId, err := base62.Random(20)
+	if err != nil {
+		return nil, err
+	}
+	return &TestRental{
+		StoreTestRental: &StoreTestRental{
+			PublicId: publicId,
+		},
+	}, nil
+}
+func (r *TestRental) TableName() string {
+	if r.table != "" {
+		return r.table
+	}
+
+	return "db_test_rental"
+}
+func (r *TestRental) SetTableName(name string) {
+	if name != "" {
+		r.table = name
+	}
+}
+
+// Scan supports Timestamps for oplogs
+func (ts *Timestamp) Scan(value interface{}) error {
+	switch t := value.(type) {
+	case time.Time:
+		var err error
+		ts.Timestamp, err = ptypes.TimestampProto(t) // google proto version
+		if err != nil {
+			return fmt.Errorf("error converting the timestamp: %w", err)
+		}
+	default:
+		return errors.New("Not a protobuf Timestamp")
+	}
+	return nil
+}
+
+// Value supports Timestamps for oplogs
+func (ts *Timestamp) Value() (driver.Value, error) {
+	if ts == nil {
+		return nil, nil
+	}
+	return ptypes.Timestamp(ts.Timestamp)
+}

internal/db/db_test/db_test.pb.go

@@ -0,0 +1,592 @@
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// 	protoc-gen-go v1.21.0
+// 	protoc        v3.11.4
+// source: controller/storage/db/db_test/v1/db_test.proto
+
+// define a test proto package for the internal/db package.  These protos
+// are only used for unit tests and are not part of the rest of the watchtower
+// domain model
+
+package db_test
+
+import (
+	proto "github.com/golang/protobuf/proto"
+	timestamp "github.com/golang/protobuf/ptypes/timestamp"
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+	reflect "reflect"
+	sync "sync"
+)
+
+const (
+	// Verify that this generated code is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
+	// Verify that runtime/protoimpl is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
+)
+
+// This is a compile-time assertion that a sufficiently up-to-date version
+// of the legacy proto package is being used.
+const _ = proto.ProtoPackageIsVersion4
+
+// Timestamp for oplog messages
+type Timestamp struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Timestamp *timestamp.Timestamp `protobuf:"bytes,1,opt,name=timestamp,proto3" json:"timestamp,omitempty"`
+}
+
+func (x *Timestamp) Reset() {
+	*x = Timestamp{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_controller_storage_db_db_test_v1_db_test_proto_msgTypes[0]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *Timestamp) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*Timestamp) ProtoMessage() {}
+
+func (x *Timestamp) ProtoReflect() protoreflect.Message {
+	mi := &file_controller_storage_db_db_test_v1_db_test_proto_msgTypes[0]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use Timestamp.ProtoReflect.Descriptor instead.
+func (*Timestamp) Descriptor() ([]byte, []int) {
+	return file_controller_storage_db_db_test_v1_db_test_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *Timestamp) GetTimestamp() *timestamp.Timestamp {
+	if x != nil {
+		return x.Timestamp
+	}
+	return nil
+}
+
+// TestUser for gorm test user model
+type StoreTestUser struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	// @inject_tag: gorm:"primary_key"
+	Id uint32 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty" gorm:"primary_key"`
+	// create_time from the RDBMS
+	// @inject_tag: `gorm:"default:CURRENT_TIMESTAMP"`
+	CreateTime *Timestamp `protobuf:"bytes,2,opt,name=create_time,json=createTime,proto3" json:"create_time,omitempty" gorm:"default:CURRENT_TIMESTAMP"`
+	// update_time from the RDBMS
+	// @inject_tag: `gorm:"default:CURRENT_TIMESTAMP"`
+	UpdateTime *Timestamp `protobuf:"bytes,3,opt,name=update_time,json=updateTime,proto3" json:"update_time,omitempty" gorm:"default:CURRENT_TIMESTAMP"`
+	// public_id is the used to access the AssignablScope via an API
+	PublicId string `protobuf:"bytes,4,opt,name=public_id,json=publicId,proto3" json:"public_id,omitempty"`
+	// FriendlyName is the optional friendly name used to
+	// access the Scope via an API
+	// @inject_tag: `gorm:"default:null"`
+	FriendlyName string `protobuf:"bytes,5,opt,name=friendly_name,json=friendlyName,proto3" json:"friendly_name,omitempty" gorm:"default:null"`
+	Name         string `protobuf:"bytes,6,opt,name=name,proto3" json:"name,omitempty"`
+	PhoneNumber  string `protobuf:"bytes,7,opt,name=phone_number,json=phoneNumber,proto3" json:"phone_number,omitempty"`
+	Email        string `protobuf:"bytes,8,opt,name=email,proto3" json:"email,omitempty"`
+}
+
+func (x *StoreTestUser) Reset() {
+	*x = StoreTestUser{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_controller_storage_db_db_test_v1_db_test_proto_msgTypes[1]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *StoreTestUser) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*StoreTestUser) ProtoMessage() {}
+
+func (x *StoreTestUser) ProtoReflect() protoreflect.Message {
+	mi := &file_controller_storage_db_db_test_v1_db_test_proto_msgTypes[1]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use StoreTestUser.ProtoReflect.Descriptor instead.
+func (*StoreTestUser) Descriptor() ([]byte, []int) {
+	return file_controller_storage_db_db_test_v1_db_test_proto_rawDescGZIP(), []int{1}
+}
+
+func (x *StoreTestUser) GetId() uint32 {
+	if x != nil {
+		return x.Id
+	}
+	return 0
+}
+
+func (x *StoreTestUser) GetCreateTime() *Timestamp {
+	if x != nil {
+		return x.CreateTime
+	}
+	return nil
+}
+
+func (x *StoreTestUser) GetUpdateTime() *Timestamp {
+	if x != nil {
+		return x.UpdateTime
+	}
+	return nil
+}
+
+func (x *StoreTestUser) GetPublicId() string {
+	if x != nil {
+		return x.PublicId
+	}
+	return ""
+}
+
+func (x *StoreTestUser) GetFriendlyName() string {
+	if x != nil {
+		return x.FriendlyName
+	}
+	return ""
+}
+
+func (x *StoreTestUser) GetName() string {
+	if x != nil {
+		return x.Name
+	}
+	return ""
+}
+
+func (x *StoreTestUser) GetPhoneNumber() string {
+	if x != nil {
+		return x.PhoneNumber
+	}
+	return ""
+}
+
+func (x *StoreTestUser) GetEmail() string {
+	if x != nil {
+		return x.Email
+	}
+	return ""
+}
+
+// TestCar for gorm test car model
+type StoreTestCar struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	// @inject_tag: gorm:"primary_key"
+	Id uint32 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty" gorm:"primary_key"`
+	// create_time from the RDBMS
+	// @inject_tag: `gorm:"default:CURRENT_TIMESTAMP"`
+	CreateTime *Timestamp `protobuf:"bytes,2,opt,name=create_time,json=createTime,proto3" json:"create_time,omitempty" gorm:"default:CURRENT_TIMESTAMP"`
+	// update_time from the RDBMS
+	// @inject_tag: `gorm:"default:CURRENT_TIMESTAMP"`
+	UpdateTime *Timestamp `protobuf:"bytes,3,opt,name=update_time,json=updateTime,proto3" json:"update_time,omitempty" gorm:"default:CURRENT_TIMESTAMP"`
+	// public_id is the used to access the AssignablScope via an API
+	PublicId string `protobuf:"bytes,4,opt,name=public_id,json=publicId,proto3" json:"public_id,omitempty"`
+	// FriendlyName is the optional friendly name used to
+	// access the Scope via an API
+	// @inject_tag: `gorm:"default:null"`
+	FriendlyName string `protobuf:"bytes,5,opt,name=friendly_name,json=friendlyName,proto3" json:"friendly_name,omitempty" gorm:"default:null"`
+	Model        string `protobuf:"bytes,6,opt,name=model,proto3" json:"model,omitempty"`
+	Mpg          int32  `protobuf:"varint,7,opt,name=mpg,proto3" json:"mpg,omitempty"`
+}
+
+func (x *StoreTestCar) Reset() {
+	*x = StoreTestCar{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_controller_storage_db_db_test_v1_db_test_proto_msgTypes[2]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *StoreTestCar) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*StoreTestCar) ProtoMessage() {}
+
+func (x *StoreTestCar) ProtoReflect() protoreflect.Message {
+	mi := &file_controller_storage_db_db_test_v1_db_test_proto_msgTypes[2]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use StoreTestCar.ProtoReflect.Descriptor instead.
+func (*StoreTestCar) Descriptor() ([]byte, []int) {
+	return file_controller_storage_db_db_test_v1_db_test_proto_rawDescGZIP(), []int{2}
+}
+
+func (x *StoreTestCar) GetId() uint32 {
+	if x != nil {
+		return x.Id
+	}
+	return 0
+}
+
+func (x *StoreTestCar) GetCreateTime() *Timestamp {
+	if x != nil {
+		return x.CreateTime
+	}
+	return nil
+}
+
+func (x *StoreTestCar) GetUpdateTime() *Timestamp {
+	if x != nil {
+		return x.UpdateTime
+	}
+	return nil
+}
+
+func (x *StoreTestCar) GetPublicId() string {
+	if x != nil {
+		return x.PublicId
+	}
+	return ""
+}
+
+func (x *StoreTestCar) GetFriendlyName() string {
+	if x != nil {
+		return x.FriendlyName
+	}
+	return ""
+}
+
+func (x *StoreTestCar) GetModel() string {
+	if x != nil {
+		return x.Model
+	}
+	return ""
+}
+
+func (x *StoreTestCar) GetMpg() int32 {
+	if x != nil {
+		return x.Mpg
+	}
+	return 0
+}
+
+// TestRental for gorm test rental model
+type StoreTestRental struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	// @inject_tag: gorm:"primary_key"
+	Id uint32 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty" gorm:"primary_key"`
+	// create_time from the RDBMS
+	// @inject_tag: `gorm:"default:CURRENT_TIMESTAMP"`
+	CreateTime *Timestamp `protobuf:"bytes,2,opt,name=create_time,json=createTime,proto3" json:"create_time,omitempty" gorm:"default:CURRENT_TIMESTAMP"`
+	// update_time from the RDBMS
+	// @inject_tag: `gorm:"default:CURRENT_TIMESTAMP"`
+	UpdateTime *Timestamp `protobuf:"bytes,3,opt,name=update_time,json=updateTime,proto3" json:"update_time,omitempty" gorm:"default:CURRENT_TIMESTAMP"`
+	// public_id is the used to access the AssignablScope via an API
+	PublicId string `protobuf:"bytes,4,opt,name=public_id,json=publicId,proto3" json:"public_id,omitempty"`
+	// FriendlyName is the optional friendly name used to
+	// access the Scope via an API
+	// @inject_tag: `gorm:"default:null"`
+	FriendlyName string `protobuf:"bytes,5,opt,name=friendly_name,json=friendlyName,proto3" json:"friendly_name,omitempty" gorm:"default:null"`
+	UserId       uint32 `protobuf:"varint,6,opt,name=user_id,json=userId,proto3" json:"user_id,omitempty"`
+	CarId        uint32 `protobuf:"varint,7,opt,name=car_id,json=carId,proto3" json:"car_id,omitempty"`
+}
+
+func (x *StoreTestRental) Reset() {
+	*x = StoreTestRental{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_controller_storage_db_db_test_v1_db_test_proto_msgTypes[3]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *StoreTestRental) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*StoreTestRental) ProtoMessage() {}
+
+func (x *StoreTestRental) ProtoReflect() protoreflect.Message {
+	mi := &file_controller_storage_db_db_test_v1_db_test_proto_msgTypes[3]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use StoreTestRental.ProtoReflect.Descriptor instead.
+func (*StoreTestRental) Descriptor() ([]byte, []int) {
+	return file_controller_storage_db_db_test_v1_db_test_proto_rawDescGZIP(), []int{3}
+}
+
+func (x *StoreTestRental) GetId() uint32 {
+	if x != nil {
+		return x.Id
+	}
+	return 0
+}
+
+func (x *StoreTestRental) GetCreateTime() *Timestamp {
+	if x != nil {
+		return x.CreateTime
+	}
+	return nil
+}
+
+func (x *StoreTestRental) GetUpdateTime() *Timestamp {
+	if x != nil {
+		return x.UpdateTime
+	}
+	return nil
+}
+
+func (x *StoreTestRental) GetPublicId() string {
+	if x != nil {
+		return x.PublicId
+	}
+	return ""
+}
+
+func (x *StoreTestRental) GetFriendlyName() string {
+	if x != nil {
+		return x.FriendlyName
+	}
+	return ""
+}
+
+func (x *StoreTestRental) GetUserId() uint32 {
+	if x != nil {
+		return x.UserId
+	}
+	return 0
+}
+
+func (x *StoreTestRental) GetCarId() uint32 {
+	if x != nil {
+		return x.CarId
+	}
+	return 0
+}
+
+var File_controller_storage_db_db_test_v1_db_test_proto protoreflect.FileDescriptor
+
+var file_controller_storage_db_db_test_v1_db_test_proto_rawDesc = []byte{
+	0x0a, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2f, 0x73, 0x74, 0x6f,
+	0x72, 0x61, 0x67, 0x65, 0x2f, 0x64, 0x62, 0x2f, 0x64, 0x62, 0x5f, 0x74, 0x65, 0x73, 0x74, 0x2f,
+	0x76, 0x31, 0x2f, 0x64, 0x62, 0x5f, 0x74, 0x65, 0x73, 0x74, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+	0x12, 0x20, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f,
+	0x72, 0x61, 0x67, 0x65, 0x2e, 0x64, 0x62, 0x2e, 0x64, 0x62, 0x5f, 0x74, 0x65, 0x73, 0x74, 0x2e,
+	0x76, 0x31, 0x1a, 0x1f, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+	0x62, 0x75, 0x66, 0x2f, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x70, 0x72,
+	0x6f, 0x74, 0x6f, 0x22, 0x45, 0x0a, 0x09, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70,
+	0x12, 0x38, 0x0a, 0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x18, 0x01, 0x20,
+	0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f,
+	0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52,
+	0x09, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x22, 0xca, 0x02, 0x0a, 0x0d, 0x53,
+	0x74, 0x6f, 0x72, 0x65, 0x54, 0x65, 0x73, 0x74, 0x55, 0x73, 0x65, 0x72, 0x12, 0x0e, 0x0a, 0x02,
+	0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x02, 0x69, 0x64, 0x12, 0x4c, 0x0a, 0x0b,
+	0x63, 0x72, 0x65, 0x61, 0x74, 0x65, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28,
+	0x0b, 0x32, 0x2b, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73,
+	0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x64, 0x62, 0x2e, 0x64, 0x62, 0x5f, 0x74, 0x65, 0x73,
+	0x74, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0a,
+	0x63, 0x72, 0x65, 0x61, 0x74, 0x65, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x4c, 0x0a, 0x0b, 0x75, 0x70,
+	0x64, 0x61, 0x74, 0x65, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32,
+	0x2b, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f,
+	0x72, 0x61, 0x67, 0x65, 0x2e, 0x64, 0x62, 0x2e, 0x64, 0x62, 0x5f, 0x74, 0x65, 0x73, 0x74, 0x2e,
+	0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0a, 0x75, 0x70,
+	0x64, 0x61, 0x74, 0x65, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x1b, 0x0a, 0x09, 0x70, 0x75, 0x62, 0x6c,
+	0x69, 0x63, 0x5f, 0x69, 0x64, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x70, 0x75, 0x62,
+	0x6c, 0x69, 0x63, 0x49, 0x64, 0x12, 0x23, 0x0a, 0x0d, 0x66, 0x72, 0x69, 0x65, 0x6e, 0x64, 0x6c,
+	0x79, 0x5f, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x66, 0x72,
+	0x69, 0x65, 0x6e, 0x64, 0x6c, 0x79, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61,
+	0x6d, 0x65, 0x18, 0x06, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x21,
+	0x0a, 0x0c, 0x70, 0x68, 0x6f, 0x6e, 0x65, 0x5f, 0x6e, 0x75, 0x6d, 0x62, 0x65, 0x72, 0x18, 0x07,
+	0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x70, 0x68, 0x6f, 0x6e, 0x65, 0x4e, 0x75, 0x6d, 0x62, 0x65,
+	0x72, 0x12, 0x14, 0x0a, 0x05, 0x65, 0x6d, 0x61, 0x69, 0x6c, 0x18, 0x08, 0x20, 0x01, 0x28, 0x09,
+	0x52, 0x05, 0x65, 0x6d, 0x61, 0x69, 0x6c, 0x22, 0xa4, 0x02, 0x0a, 0x0c, 0x53, 0x74, 0x6f, 0x72,
+	0x65, 0x54, 0x65, 0x73, 0x74, 0x43, 0x61, 0x72, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01,
+	0x20, 0x01, 0x28, 0x0d, 0x52, 0x02, 0x69, 0x64, 0x12, 0x4c, 0x0a, 0x0b, 0x63, 0x72, 0x65, 0x61,
+	0x74, 0x65, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2b, 0x2e,
+	0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61,
+	0x67, 0x65, 0x2e, 0x64, 0x62, 0x2e, 0x64, 0x62, 0x5f, 0x74, 0x65, 0x73, 0x74, 0x2e, 0x76, 0x31,
+	0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0a, 0x63, 0x72, 0x65, 0x61,
+	0x74, 0x65, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x4c, 0x0a, 0x0b, 0x75, 0x70, 0x64, 0x61, 0x74, 0x65,
+	0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2b, 0x2e, 0x63, 0x6f,
+	0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65,
+	0x2e, 0x64, 0x62, 0x2e, 0x64, 0x62, 0x5f, 0x74, 0x65, 0x73, 0x74, 0x2e, 0x76, 0x31, 0x2e, 0x54,
+	0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0a, 0x75, 0x70, 0x64, 0x61, 0x74, 0x65,
+	0x54, 0x69, 0x6d, 0x65, 0x12, 0x1b, 0x0a, 0x09, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x5f, 0x69,
+	0x64, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x49,
+	0x64, 0x12, 0x23, 0x0a, 0x0d, 0x66, 0x72, 0x69, 0x65, 0x6e, 0x64, 0x6c, 0x79, 0x5f, 0x6e, 0x61,
+	0x6d, 0x65, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x66, 0x72, 0x69, 0x65, 0x6e, 0x64,
+	0x6c, 0x79, 0x4e, 0x61, 0x6d, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x6d, 0x6f, 0x64, 0x65, 0x6c, 0x18,
+	0x06, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x6d, 0x6f, 0x64, 0x65, 0x6c, 0x12, 0x10, 0x0a, 0x03,
+	0x6d, 0x70, 0x67, 0x18, 0x07, 0x20, 0x01, 0x28, 0x05, 0x52, 0x03, 0x6d, 0x70, 0x67, 0x22, 0xaf,
+	0x02, 0x0a, 0x0f, 0x53, 0x74, 0x6f, 0x72, 0x65, 0x54, 0x65, 0x73, 0x74, 0x52, 0x65, 0x6e, 0x74,
+	0x61, 0x6c, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x02,
+	0x69, 0x64, 0x12, 0x4c, 0x0a, 0x0b, 0x63, 0x72, 0x65, 0x61, 0x74, 0x65, 0x5f, 0x74, 0x69, 0x6d,
+	0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2b, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f,
+	0x6c, 0x6c, 0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x64, 0x62, 0x2e,
+	0x64, 0x62, 0x5f, 0x74, 0x65, 0x73, 0x74, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73,
+	0x74, 0x61, 0x6d, 0x70, 0x52, 0x0a, 0x63, 0x72, 0x65, 0x61, 0x74, 0x65, 0x54, 0x69, 0x6d, 0x65,
+	0x12, 0x4c, 0x0a, 0x0b, 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x5f, 0x74, 0x69, 0x6d, 0x65, 0x18,
+	0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x2b, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x6c,
+	0x65, 0x72, 0x2e, 0x73, 0x74, 0x6f, 0x72, 0x61, 0x67, 0x65, 0x2e, 0x64, 0x62, 0x2e, 0x64, 0x62,
+	0x5f, 0x74, 0x65, 0x73, 0x74, 0x2e, 0x76, 0x31, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61,
+	0x6d, 0x70, 0x52, 0x0a, 0x75, 0x70, 0x64, 0x61, 0x74, 0x65, 0x54, 0x69, 0x6d, 0x65, 0x12, 0x1b,
+	0x0a, 0x09, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x5f, 0x69, 0x64, 0x18, 0x04, 0x20, 0x01, 0x28,
+	0x09, 0x52, 0x08, 0x70, 0x75, 0x62, 0x6c, 0x69, 0x63, 0x49, 0x64, 0x12, 0x23, 0x0a, 0x0d, 0x66,
+	0x72, 0x69, 0x65, 0x6e, 0x64, 0x6c, 0x79, 0x5f, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x05, 0x20, 0x01,
+	0x28, 0x09, 0x52, 0x0c, 0x66, 0x72, 0x69, 0x65, 0x6e, 0x64, 0x6c, 0x79, 0x4e, 0x61, 0x6d, 0x65,
+	0x12, 0x17, 0x0a, 0x07, 0x75, 0x73, 0x65, 0x72, 0x5f, 0x69, 0x64, 0x18, 0x06, 0x20, 0x01, 0x28,
+	0x0d, 0x52, 0x06, 0x75, 0x73, 0x65, 0x72, 0x49, 0x64, 0x12, 0x15, 0x0a, 0x06, 0x63, 0x61, 0x72,
+	0x5f, 0x69, 0x64, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x05, 0x63, 0x61, 0x72, 0x49, 0x64,
+	0x42, 0x3d, 0x5a, 0x3b, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68,
+	0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f, 0x77, 0x61, 0x74, 0x63, 0x68, 0x74, 0x6f,
+	0x77, 0x65, 0x72, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x64, 0x62, 0x2f,
+	0x64, 0x62, 0x5f, 0x74, 0x65, 0x73, 0x74, 0x3b, 0x64, 0x62, 0x5f, 0x74, 0x65, 0x73, 0x74, 0x62,
+	0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+}
+
+var (
+	file_controller_storage_db_db_test_v1_db_test_proto_rawDescOnce sync.Once
+	file_controller_storage_db_db_test_v1_db_test_proto_rawDescData = file_controller_storage_db_db_test_v1_db_test_proto_rawDesc
+)
+
+func file_controller_storage_db_db_test_v1_db_test_proto_rawDescGZIP() []byte {
+	file_controller_storage_db_db_test_v1_db_test_proto_rawDescOnce.Do(func() {
+		file_controller_storage_db_db_test_v1_db_test_proto_rawDescData = protoimpl.X.CompressGZIP(file_controller_storage_db_db_test_v1_db_test_proto_rawDescData)
+	})
+	return file_controller_storage_db_db_test_v1_db_test_proto_rawDescData
+}
+
+var file_controller_storage_db_db_test_v1_db_test_proto_msgTypes = make([]protoimpl.MessageInfo, 4)
+var file_controller_storage_db_db_test_v1_db_test_proto_goTypes = []interface{}{
+	(*Timestamp)(nil),           // 0: controller.storage.db.db_test.v1.Timestamp
+	(*StoreTestUser)(nil),       // 1: controller.storage.db.db_test.v1.StoreTestUser
+	(*StoreTestCar)(nil),        // 2: controller.storage.db.db_test.v1.StoreTestCar
+	(*StoreTestRental)(nil),     // 3: controller.storage.db.db_test.v1.StoreTestRental
+	(*timestamp.Timestamp)(nil), // 4: google.protobuf.Timestamp
+}
+var file_controller_storage_db_db_test_v1_db_test_proto_depIdxs = []int32{
+	4, // 0: controller.storage.db.db_test.v1.Timestamp.timestamp:type_name -> google.protobuf.Timestamp
+	0, // 1: controller.storage.db.db_test.v1.StoreTestUser.create_time:type_name -> controller.storage.db.db_test.v1.Timestamp
+	0, // 2: controller.storage.db.db_test.v1.StoreTestUser.update_time:type_name -> controller.storage.db.db_test.v1.Timestamp
+	0, // 3: controller.storage.db.db_test.v1.StoreTestCar.create_time:type_name -> controller.storage.db.db_test.v1.Timestamp
+	0, // 4: controller.storage.db.db_test.v1.StoreTestCar.update_time:type_name -> controller.storage.db.db_test.v1.Timestamp
+	0, // 5: controller.storage.db.db_test.v1.StoreTestRental.create_time:type_name -> controller.storage.db.db_test.v1.Timestamp
+	0, // 6: controller.storage.db.db_test.v1.StoreTestRental.update_time:type_name -> controller.storage.db.db_test.v1.Timestamp
+	7, // [7:7] is the sub-list for method output_type
+	7, // [7:7] is the sub-list for method input_type
+	7, // [7:7] is the sub-list for extension type_name
+	7, // [7:7] is the sub-list for extension extendee
+	0, // [0:7] is the sub-list for field type_name
+}
+
+func init() { file_controller_storage_db_db_test_v1_db_test_proto_init() }
+func file_controller_storage_db_db_test_v1_db_test_proto_init() {
+	if File_controller_storage_db_db_test_v1_db_test_proto != nil {
+		return
+	}
+	if !protoimpl.UnsafeEnabled {
+		file_controller_storage_db_db_test_v1_db_test_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*Timestamp); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_controller_storage_db_db_test_v1_db_test_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*StoreTestUser); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_controller_storage_db_db_test_v1_db_test_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*StoreTestCar); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_controller_storage_db_db_test_v1_db_test_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*StoreTestRental); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+	}
+	type x struct{}
+	out := protoimpl.TypeBuilder{
+		File: protoimpl.DescBuilder{
+			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+			RawDescriptor: file_controller_storage_db_db_test_v1_db_test_proto_rawDesc,
+			NumEnums:      0,
+			NumMessages:   4,
+			NumExtensions: 0,
+			NumServices:   0,
+		},
+		GoTypes:           file_controller_storage_db_db_test_v1_db_test_proto_goTypes,
+		DependencyIndexes: file_controller_storage_db_db_test_v1_db_test_proto_depIdxs,
+		MessageInfos:      file_controller_storage_db_db_test_v1_db_test_proto_msgTypes,
+	}.Build()
+	File_controller_storage_db_db_test_v1_db_test_proto = out.File
+	file_controller_storage_db_db_test_v1_db_test_proto_rawDesc = nil
+	file_controller_storage_db_db_test_v1_db_test_proto_goTypes = nil
+	file_controller_storage_db_db_test_v1_db_test_proto_depIdxs = nil
+}

internal/db/migrations/postgres/01_oplog.up.sql

@@ -0,0 +1,39 @@
+CREATE TABLE if not exists oplog_entry (
+  id bigint generated always as identity primary key,
+  create_time timestamp with time zone default current_timestamp,
+  update_time timestamp with time zone default current_timestamp,
+  version text NOT NULL,
+  aggregate_name text NOT NULL,
+  "data" bytea NOT NULL
+);
+CREATE TABLE if not exists oplog_ticket (
+  id bigint generated always as identity primary key,
+  create_time timestamp with time zone default current_timestamp,
+  update_time timestamp with time zone default current_timestamp,
+  "name" text NOT NULL UNIQUE,
+  "version" bigint NOT NULL
+);
+CREATE TABLE if not exists oplog_metadata (
+  id bigint generated always as identity primary key,
+  create_time timestamp with time zone default current_timestamp,
+  entry_id bigint NOT NULL REFERENCES oplog_entry(id) ON DELETE CASCADE ON UPDATE CASCADE,
+  "key" text NOT NULL,
+  value text NULL
+);
+create index if not exists idx_oplog_metatadata_key on oplog_metadata(key);
+create index if not exists idx_oplog_metatadata_value on oplog_metadata(value);
+INSERT INTO oplog_ticket (name, version)
+values
+  ('default', 1),
+  ('iam_scope', 1),
+  ('iam_user', 1),
+  ('iam_auth_method', 1),
+  ('iam_group', 1),
+  ('iam_group_member_user', 1),
+  ('iam_role', 1),
+  ('iam_role_grant', 1),
+  ('iam_role_group', 1),
+  ('iam_role_user', 1),
+  ('db_test_user', 1),
+  ('db_test_car', 1),
+  ('db_test_rental', 1);

internal/db/migrations/postgres/03_db.down.sql

@@ -0,0 +1,3 @@
+drop table if exists db_test_user;
+drop table if exists db_test_car;
+drop table if exists db_test_rental;

internal/db/migrations/postgres/03_db.up.sql

@@ -0,0 +1,31 @@
+-- create test tables used in the unit tests for the internal/db package
+-- these tables (db_test_user, db_test_car, db_test_rental) are not part
+-- of the Watchtower domain model... they are simply used for testing the internal/db package
+CREATE TABLE if not exists db_test_user (
+  id bigint generated always as identity primary key,
+  create_time timestamp with time zone default current_timestamp,
+  update_time timestamp with time zone default current_timestamp,
+  public_id text NOT NULL UNIQUE,
+  friendly_name text UNIQUE,
+  name text,
+  phone_number text,
+  email text
+);
+CREATE TABLE if not exists db_test_car (
+  id bigint generated always as identity primary key,
+  create_time timestamp with time zone default current_timestamp,
+  update_time timestamp with time zone default current_timestamp,
+  public_id text NOT NULL UNIQUE,
+  friendly_name text UNIQUE,
+  model text,
+  mpg smallint
+);
+CREATE TABLE if not exists db_test_rental (
+  id bigint generated always as identity primary key,
+  create_time timestamp with time zone default current_timestamp,
+  update_time timestamp with time zone default current_timestamp,
+  public_id text NOT NULL UNIQUE,
+  friendly_name text UNIQUE,
+  user_id bigint not null REFERENCES db_test_user(id),
+  car_id bigint not null REFERENCES db_test_car(id)
+);

internal/db/option.go

@@ -0,0 +1,68 @@
+package db
+
+import (
+	wrapping "github.com/hashicorp/go-kms-wrapping"
+	"github.com/hashicorp/watchtower/internal/oplog"
+)
+
+// getOpts - iterate the inbound Options and return a struct
+func getOpts(opt ...Option) options {
+	opts := getDefaultOptions()
+	for _, o := range opt {
+		o(&opts)
+	}
+	return opts
+}
+
+// Option - how Options are passed as arguments
+type Option func(*options)
+
+// options = how options are represented
+type options struct {
+	withOplog  bool
+	oplogOpts  oplogOpts
+	withDebug  bool
+	withLookup bool
+}
+
+type oplogOpts struct {
+	wrapper  wrapping.Wrapper
+	metadata oplog.Metadata
+}
+
+func getDefaultOptions() options {
+	return options{
+		withOplog: false,
+		oplogOpts: oplogOpts{
+			wrapper:  nil,
+			metadata: oplog.Metadata{},
+		},
+		withDebug:  false,
+		withLookup: false,
+	}
+}
+
+// WithLookup enables a lookup
+func WithLookup(enable bool) Option {
+	return func(o *options) {
+		o.withLookup = enable
+	}
+}
+
+// WithDebug enables debug
+func WithDebug(enable bool) Option {
+	return func(o *options) {
+		o.withDebug = enable
+	}
+}
+
+// WithOplog provides an option to write an oplog entry
+func WithOplog(wrapper wrapping.Wrapper, md oplog.Metadata) Option {
+	return func(o *options) {
+		o.withOplog = true
+		o.oplogOpts = oplogOpts{
+			wrapper:  wrapper,
+			metadata: md,
+		}
+	}
+}

internal/db/option_test.go

@@ -0,0 +1,65 @@
+package db
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/hashicorp/watchtower/internal/oplog"
+	"github.com/stretchr/testify/assert"
+)
+
+// Test_getOpts provides unit tests for GetOpts and all the options
+func Test_getOpts(t *testing.T) {
+	assert := assert.New(t)
+	t.Parallel()
+	t.Run("WithOplog", func(t *testing.T) {
+		// test default of false
+		opts := getOpts()
+		testOpts := getDefaultOptions()
+		testOpts.withOplog = false
+		assert.True(reflect.DeepEqual(opts, testOpts))
+
+		wrapper := TestWrapper(t)
+		md := oplog.Metadata{
+			"key-only":   nil,
+			"deployment": []string{"amex"},
+			"project":    []string{"central-info-systems", "local-info-systems"},
+		}
+		// try setting to true
+		opts = getOpts(WithOplog(wrapper, md))
+		testOpts = getDefaultOptions()
+		testOpts.withOplog = true
+		testOpts.oplogOpts = oplogOpts{
+			wrapper:  wrapper,
+			metadata: md,
+		}
+		assert.True(reflect.DeepEqual(opts, testOpts))
+	})
+	t.Run("WithLookup", func(t *testing.T) {
+		// test default of true
+		opts := getOpts()
+		testOpts := getDefaultOptions()
+		testOpts.withLookup = false
+		assert.True(reflect.DeepEqual(opts, testOpts))
+
+		// try setting to false
+		opts = getOpts(WithLookup(true))
+		testOpts = getDefaultOptions()
+		testOpts.withLookup = true
+		assert.True(reflect.DeepEqual(opts, testOpts))
+
+	})
+	t.Run("WithDebug", func(t *testing.T) {
+		// test default of false
+		opts := getOpts()
+		testOpts := getDefaultOptions()
+		testOpts.withDebug = false
+		assert.True(reflect.DeepEqual(opts, testOpts))
+
+		// try setting to true
+		opts = getOpts(WithDebug(true))
+		testOpts = getDefaultOptions()
+		testOpts.withDebug = true
+		assert.True(reflect.DeepEqual(opts, testOpts))
+	})
+}

internal/db/read_writer.go

@@ -0,0 +1,444 @@
+package db
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"reflect"
+	"strings"
+	"time"
+
+	"github.com/hashicorp/watchtower/internal/oplog"
+	"github.com/jinzhu/gorm"
+	"google.golang.org/protobuf/proto"
+)
+
+// Reader interface defines lookups/searching for resources
+type Reader interface {
+	// LookupByFriendlyName will lookup resource by its friendly_name which must be unique
+	LookupByFriendlyName(ctx context.Context, resource ResourceFriendlyNamer, opt ...Option) error
+
+	// LookupByPublicId will lookup resource by its public_id which must be unique
+	LookupByPublicId(ctx context.Context, resource ResourcePublicIder, opt ...Option) error
+
+	// LookupWhere will lookup and return the first resource using a where clause with parameters
+	LookupWhere(ctx context.Context, resource interface{}, where string, args ...interface{}) error
+
+	// SearchWhere will search for all the resources it can find using a where clause with parameters
+	SearchWhere(ctx context.Context, resources interface{}, where string, args ...interface{}) error
+
+	// ScanRows will scan sql rows into the interface provided
+	ScanRows(rows *sql.Rows, result interface{}) error
+
+	// DB returns the sql.DB
+	DB() (*sql.DB, error)
+}
+
+// Writer interface defines create, update and retryable transaction handlers
+type Writer interface {
+	// DoTx will wrap the TxHandler in a retryable transaction
+	DoTx(ctx context.Context, retries uint, backOff Backoff, Handler TxHandler) (RetryInfo, error)
+
+	// Update an object in the db, if there's a fieldMask then only the field_mask.proto paths are updated, otherwise
+	// it will send every field to the DB.  options: WithOplog
+	// the caller is responsible for the transaction life cycle of the writer
+	// and if an error is returned the caller must decide what to do with
+	// the transaction, which almost always should be to rollback.
+	Update(ctx context.Context, i interface{}, fieldMaskPaths []string, opt ...Option) error
+
+	// Create an object in the db with options: WithOplog
+	// the caller is responsible for the transaction life cycle of the writer
+	// and if an error is returned the caller must decide what to do with
+	// the transaction, which almost always should be to rollback.
+	Create(ctx context.Context, i interface{}, opt ...Option) error
+
+	// Delete an object in the db with options: WithOplog
+	// the caller is responsible for the transaction life cycle of the writer
+	// and if an error is returned the caller must decide what to do with
+	// the transaction, which almost always should be to rollback.
+	Delete(ctx context.Context, i interface{}, opt ...Option) error
+
+	// DB returns the sql.DB
+	DB() (*sql.DB, error)
+}
+
+// RetryInfo provides information on the retries of a transaction
+type RetryInfo struct {
+	Retries int
+	Backoff time.Duration
+}
+
+// TxHandler defines a handler for a func that writes a transaction for use with DoTx
+type TxHandler func(Writer) error
+
+// ResourcePublicIder defines an interface that LookupByPublicId() can use to get the resource's public id
+type ResourcePublicIder interface {
+	GetPublicId() string
+}
+
+// ResourceFriendlyNamer defines an interface that LookupByFriendlyName() can use to get the resource's friendly name
+type ResourceFriendlyNamer interface {
+	GetFriendlyName() string
+}
+
+type OpType int
+
+const (
+	UnknownOp OpType = 0
+	CreateOp  OpType = 1
+	UpdateOp  OpType = 2
+	DeleteOp  OpType = 3
+)
+
+// VetForWriter provides an interface that Create and Update can use to vet the resource before sending it to the db
+type VetForWriter interface {
+	VetForWrite(ctx context.Context, r Reader, opType OpType) error
+}
+
+// GormReadWriter uses a gorm DB connection for read/write
+type GormReadWriter struct {
+	Tx *gorm.DB
+}
+
+// ensure that GroupRole implements the interfaces of: Resource, ClonableResource, AssignedRole and db.VetForWriter
+var _ Reader = (*GormReadWriter)(nil)
+var _ Writer = (*GormReadWriter)(nil)
+
+// DB returns the sql.DB
+func (rw *GormReadWriter) DB() (*sql.DB, error) {
+	if rw.Tx == nil {
+		return nil, errors.New("create tx is nil for db")
+	}
+	return rw.Tx.DB(), nil
+}
+
+// Scan rows will scan the rows into the interface
+func (rw *GormReadWriter) ScanRows(rows *sql.Rows, result interface{}) error {
+	return rw.Tx.ScanRows(rows, result)
+}
+
+// CreateConstraint will create a db constraint if it doesn't already exist
+func (w *GormReadWriter) CreateConstraint(tableName string, constraintName string, constraint string) error {
+	return w.Tx.Exec("create_constraint_if_not_exists(?, ?, ?)", tableName, constraintName, constraint).Error
+}
+
+var ErrNotResourceWithId = errors.New("not a resource with an id")
+
+func (rw *GormReadWriter) lookupAfterWrite(ctx context.Context, i interface{}, opt ...Option) error {
+	opts := getOpts(opt...)
+	withLookup := opts.withLookup
+
+	if !withLookup {
+		return nil
+	}
+	if _, ok := i.(ResourcePublicIder); ok {
+		if err := rw.LookupByPublicId(ctx, i.(ResourcePublicIder), opt...); err != nil {
+			return err
+		}
+		return nil
+	}
+	return ErrNotResourceWithId
+}
+
+// Create an object in the db with options: WithOplog and WithLookup (to force a lookup after create))
+func (rw *GormReadWriter) Create(ctx context.Context, i interface{}, opt ...Option) error {
+	if rw.Tx == nil {
+		return errors.New("create tx is nil")
+	}
+	opts := getOpts(opt...)
+	withOplog := opts.withOplog
+	withDebug := opts.withDebug
+	if withDebug {
+		rw.Tx.LogMode(true)
+		defer rw.Tx.LogMode(false)
+	}
+	if i == nil {
+		return errors.New("create interface is nil")
+	}
+	if vetter, ok := i.(VetForWriter); ok {
+		if err := vetter.VetForWrite(ctx, rw, CreateOp); err != nil {
+			return fmt.Errorf("error on Create %w", err)
+		}
+	}
+	if err := rw.Tx.Create(i).Error; err != nil {
+		return fmt.Errorf("error creating: %w", err)
+	}
+	if withOplog {
+		if err := rw.addOplog(ctx, CreateOp, opts, i); err != nil {
+			return err
+		}
+	}
+	if err := rw.lookupAfterWrite(ctx, i, opt...); err != nil {
+		return fmt.Errorf("lookup error after create: %w", err)
+	}
+	return nil
+}
+
+// Update an object in the db, if there's a fieldMask then only the field_mask.proto paths are updated, otherwise
+// it will send every field to the DB.  Update supports embedding a struct (or structPtr) one level deep for updating
+func (rw *GormReadWriter) Update(ctx context.Context, i interface{}, fieldMaskPaths []string, opt ...Option) error {
+	if rw.Tx == nil {
+		return errors.New("update tx is nil")
+	}
+	opts := getOpts(opt...)
+	withDebug := opts.withDebug
+	withOplog := opts.withOplog
+	if withDebug {
+		rw.Tx.LogMode(true)
+		defer rw.Tx.LogMode(false)
+	}
+	if i == nil {
+		return errors.New("update interface is nil")
+	}
+	if vetter, ok := i.(VetForWriter); ok {
+		if err := vetter.VetForWrite(ctx, rw, UpdateOp); err != nil {
+			return fmt.Errorf("error on Create %w", err)
+		}
+	}
+	if len(fieldMaskPaths) == 0 {
+		if err := rw.Tx.Save(i).Error; err != nil {
+			return fmt.Errorf("error updating: %w", err)
+		}
+	}
+	updateFields := map[string]interface{}{}
+
+	val := reflect.Indirect(reflect.ValueOf(i))
+	structTyp := val.Type()
+	for _, field := range fieldMaskPaths {
+		for i := 0; i < structTyp.NumField(); i++ {
+			kind := structTyp.Field(i).Type.Kind()
+			if kind == reflect.Struct || kind == reflect.Ptr {
+				embType := structTyp.Field(i).Type
+				// check if the embedded field is exported via CanInterface()
+				if val.Field(i).CanInterface() {
+					embVal := reflect.Indirect(reflect.ValueOf(val.Field(i).Interface()))
+					// if it's a ptr to a struct, then we need a few more bits before proceeding.
+					if kind == reflect.Ptr {
+						embVal = val.Field(i).Elem()
+						embType = embVal.Type()
+						if embType.Kind() != reflect.Struct {
+							continue
+						}
+					}
+					for embFieldNum := 0; embFieldNum < embType.NumField(); embFieldNum++ {
+						if strings.EqualFold(embType.Field(embFieldNum).Name, field) {
+							updateFields[field] = embVal.Field(embFieldNum).Interface()
+						}
+					}
+					continue
+				}
+			}
+			// it's not an embedded type, so check if the field name matches
+			if strings.EqualFold(structTyp.Field(i).Name, field) {
+				updateFields[field] = val.Field(i).Interface()
+			}
+		}
+	}
+	if len(updateFields) == 0 {
+		return fmt.Errorf("error no update fields matched using fieldMaskPaths: %s", fieldMaskPaths)
+	}
+	if err := rw.Tx.Model(i).Updates(updateFields).Error; err != nil {
+		return fmt.Errorf("error updating: %w", err)
+	}
+	if withOplog {
+		if err := rw.addOplog(ctx, UpdateOp, opts, i); err != nil {
+			return err
+		}
+	}
+	if err := rw.lookupAfterWrite(ctx, i, opt...); err != nil {
+		return fmt.Errorf("lookup error after update: %w", err)
+	}
+	return nil
+}
+
+// Delete an object in the db with options: WithOplog (which requires WithMetadata, WithWrapper)
+func (rw *GormReadWriter) Delete(ctx context.Context, i interface{}, opt ...Option) error {
+	if rw.Tx == nil {
+		return errors.New("delete tx is nil")
+	}
+	if i == nil {
+		return errors.New("delete interface is nil")
+	}
+	opts := getOpts(opt...)
+	withDebug := opts.withDebug
+	withOplog := opts.withOplog
+	if withDebug {
+		rw.Tx.LogMode(true)
+		defer rw.Tx.LogMode(false)
+	}
+	if err := rw.Tx.Delete(i).Error; err != nil {
+		return fmt.Errorf("error deleting: %w", err)
+	}
+	if withOplog {
+		if err := rw.addOplog(ctx, DeleteOp, opts, i); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (rw *GormReadWriter) addOplog(ctx context.Context, opType OpType, opts options, i interface{}) error {
+	oplogArgs := opts.oplogOpts
+	if oplogArgs.wrapper == nil {
+		return errors.New("error wrapper is nil for WithWrapper")
+	}
+	if len(oplogArgs.metadata) == 0 {
+		return errors.New("error no metadata for WithOplog")
+	}
+	replayable, ok := i.(oplog.ReplayableMessage)
+	if !ok {
+		return errors.New("error not a replayable message for WithOplog")
+	}
+	gdb := rw.Tx
+	withDebug := opts.withDebug
+	if withDebug {
+		gdb.LogMode(true)
+		defer gdb.LogMode(false)
+	}
+	ticketer, err := oplog.NewGormTicketer(gdb, oplog.WithAggregateNames(true))
+	if err != nil {
+		return fmt.Errorf("error getting Ticketer %w for WithOplog", err)
+	}
+	err = ticketer.InitTicket(replayable.TableName())
+	if err != nil {
+		return fmt.Errorf("error getting initializing ticket %w for WithOplog", err)
+	}
+	ticket, err := ticketer.GetTicket(replayable.TableName())
+	if err != nil {
+		return fmt.Errorf("error getting ticket %w for WithOplog", err)
+	}
+
+	entry, err := oplog.NewEntry(
+		replayable.TableName(),
+		oplogArgs.metadata,
+		oplogArgs.wrapper,
+		ticketer,
+	)
+	var entryOp oplog.OpType
+	switch opType {
+	case CreateOp:
+		entryOp = oplog.OpType_OP_TYPE_CREATE
+	case UpdateOp:
+		entryOp = oplog.OpType_OP_TYPE_UPDATE
+	case DeleteOp:
+		entryOp = oplog.OpType_OP_TYPE_DELETE
+	default:
+		return fmt.Errorf("error operation type %v is not supported", opType)
+	}
+	err = entry.WriteEntryWith(
+		ctx,
+		&oplog.GormWriter{Tx: gdb},
+		ticket,
+		&oplog.Message{Message: i.(proto.Message), TypeName: replayable.TableName(), OpType: entryOp},
+	)
+	if err != nil {
+		return fmt.Errorf("error creating oplog entry %w for WithOplog", err)
+	}
+	return nil
+}
+
+// DoTx will wrap the Handler func passed within a transaction with retries
+// you should ensure that any objects written to the db in your TxHandler are retryable, which
+// means that the object may be sent to the db several times (retried), so things like the primary key must
+// be reset before retry
+func (w *GormReadWriter) DoTx(ctx context.Context, retries uint, backOff Backoff, Handler TxHandler) (RetryInfo, error) {
+	if w.Tx == nil {
+		return RetryInfo{}, errors.New("do tx is nil")
+	}
+	info := RetryInfo{}
+	for attempts := uint(1); ; attempts++ {
+		if attempts > retries+1 {
+			return info, fmt.Errorf("Too many retries: %d of %d", attempts-1, retries+1)
+		}
+		newTx := w.Tx.BeginTx(ctx, nil)
+		if err := Handler(&GormReadWriter{newTx}); err != nil {
+			if errors.Is(err, oplog.ErrTicketAlreadyRedeemed) {
+				newTx.Rollback() // need to still handle rollback errors
+				d := backOff.Duration(attempts)
+				info.Retries++
+				info.Backoff = info.Backoff + d
+				time.Sleep(d)
+				continue
+			} else {
+				return info, err
+			}
+		} else {
+			if err := newTx.Commit().Error; err != nil {
+				if errors.Is(err, oplog.ErrTicketAlreadyRedeemed) {
+					d := backOff.Duration(attempts)
+					info.Retries++
+					info.Backoff = info.Backoff + d
+					time.Sleep(d)
+					continue
+				} else {
+					return info, err
+				}
+			}
+			break // it all worked!!!
+		}
+	}
+	return info, nil
+}
+
+// LookupByFriendlyName will lookup resource my its friendly_name which must be unique
+func (rw *GormReadWriter) LookupByFriendlyName(ctx context.Context, resource ResourceFriendlyNamer, opt ...Option) error {
+	if rw.Tx == nil {
+		return errors.New("error tx nil for lookup by friendly name")
+	}
+	opts := getOpts(opt...)
+	withDebug := opts.withDebug
+	if withDebug {
+		rw.Tx.LogMode(true)
+		defer rw.Tx.LogMode(false)
+	}
+	if reflect.ValueOf(resource).Kind() != reflect.Ptr {
+		return errors.New("error interface parameter must to be a pointer for lookup by friendly name")
+	}
+	if resource.GetFriendlyName() == "" {
+		return errors.New("error friendly name empty string for lookup by friendly name")
+	}
+	return rw.Tx.Where("friendly_name = ?", resource.GetFriendlyName()).First(resource).Error
+}
+
+// LookupByPublicId will lookup resource my its public_id which must be unique
+func (rw *GormReadWriter) LookupByPublicId(ctx context.Context, resource ResourcePublicIder, opt ...Option) error {
+	if rw.Tx == nil {
+		return errors.New("error tx nil for lookup by public id")
+	}
+	opts := getOpts(opt...)
+	withDebug := opts.withDebug
+	if withDebug {
+		rw.Tx.LogMode(true)
+		defer rw.Tx.LogMode(false)
+	}
+	if reflect.ValueOf(resource).Kind() != reflect.Ptr {
+		return errors.New("error interface parameter must to be a pointer for lookup by public id")
+	}
+	if resource.GetPublicId() == "" {
+		return errors.New("error public id empty string for lookup by public id")
+	}
+	return rw.Tx.Where("public_id = ?", resource.GetPublicId()).First(resource).Error
+}
+
+// LookupWhere will lookup the first resource using a where clause with parameters (it only returns the first one)
+func (rw *GormReadWriter) LookupWhere(ctx context.Context, resource interface{}, where string, args ...interface{}) error {
+	if rw.Tx == nil {
+		return errors.New("error tx nil for lookup by")
+	}
+	if reflect.ValueOf(resource).Kind() != reflect.Ptr {
+		return errors.New("error interface parameter must to be a pointer for lookup by")
+	}
+	return rw.Tx.Where(where, args...).First(resource).Error
+}
+
+// SearchWhere will search for all the resources it can find using a where clause with parameters
+func (rw *GormReadWriter) SearchWhere(ctx context.Context, resources interface{}, where string, args ...interface{}) error {
+	if rw.Tx == nil {
+		return errors.New("error tx nil for search by")
+	}
+	if reflect.ValueOf(resources).Kind() != reflect.Ptr {
+		return errors.New("error interface parameter must to be a pointer for search by")
+	}
+	return rw.Tx.Where(where, args...).Find(resources).Error
+}

internal/db/read_writer_test.go

@@ -0,0 +1,814 @@
+package db
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/hashicorp/go-uuid"
+	"github.com/hashicorp/watchtower/internal/db/db_test"
+	"github.com/hashicorp/watchtower/internal/oplog"
+	"github.com/hashicorp/watchtower/internal/oplog/store"
+	"github.com/jinzhu/gorm"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestGormReadWriter_Update(t *testing.T) {
+	// intentionally not run with t.Parallel so we don't need to use DoTx for the Update tests
+	cleanup, db := TestSetup(t, "migrations/postgres")
+	defer cleanup()
+	assert := assert.New(t)
+	defer db.Close()
+	t.Run("simple", func(t *testing.T) {
+		w := GormReadWriter{Tx: db}
+		id, err := uuid.GenerateUUID()
+		assert.Nil(err)
+		user, err := db_test.NewTestUser()
+		assert.Nil(err)
+		user.Name = "foo-" + id
+		err = w.Create(context.Background(), user)
+		assert.Nil(err)
+		assert.True(user.Id != 0)
+
+		foundUser, err := db_test.NewTestUser()
+		assert.Nil(err)
+		foundUser.PublicId = user.PublicId
+		err = w.LookupByPublicId(context.Background(), foundUser)
+		assert.Nil(err)
+		assert.Equal(foundUser.Id, user.Id)
+
+		user.FriendlyName = "friendly-" + id
+		err = w.Update(context.Background(), user, []string{"FriendlyName"})
+		assert.Nil(err)
+
+		err = w.LookupByPublicId(context.Background(), foundUser)
+		assert.Nil(err)
+		assert.Equal(foundUser.FriendlyName, user.FriendlyName)
+	})
+	t.Run("valid-WithOplog", func(t *testing.T) {
+		w := GormReadWriter{Tx: db}
+		id, err := uuid.GenerateUUID()
+		assert.Nil(err)
+		user, err := db_test.NewTestUser()
+		assert.Nil(err)
+		user.Name = "foo-" + id
+		err = w.Create(context.Background(), user)
+		assert.Nil(err)
+		assert.True(user.Id != 0)
+
+		foundUser, err := db_test.NewTestUser()
+		assert.Nil(err)
+		foundUser.PublicId = user.PublicId
+		err = w.LookupByPublicId(context.Background(), foundUser)
+		assert.Nil(err)
+		assert.Equal(foundUser.Id, user.Id)
+
+		user.FriendlyName = "friendly-" + id
+		err = w.Update(context.Background(), user, []string{"FriendlyName"},
+			// write oplogs for this update
+			WithOplog(
+				TestWrapper(t),
+				oplog.Metadata{
+					"key-only":           nil,
+					"deployment":         []string{"amex"},
+					"project":            []string{"central-info-systems", "local-info-systems"},
+					"resource-public-id": []string{user.GetPublicId()},
+				}),
+		)
+		assert.Nil(err)
+
+		err = w.LookupByPublicId(context.Background(), foundUser)
+		assert.Nil(err)
+		assert.Equal(foundUser.FriendlyName, user.FriendlyName)
+
+		var metadata store.Metadata
+		err = db.Where("key = ? and value = ?", "resource-public-id", user.PublicId).First(&metadata).Error
+		assert.Nil(err)
+
+		var foundEntry oplog.Entry
+		err = db.Where("id = ?", metadata.EntryId).First(&foundEntry).Error
+		assert.Nil(err)
+	})
+	t.Run("nil-tx", func(t *testing.T) {
+		w := GormReadWriter{Tx: nil}
+		id, err := uuid.GenerateUUID()
+		assert.Nil(err)
+		user, err := db_test.NewTestUser()
+		assert.Nil(err)
+		user.Name = "foo-" + id
+		err = w.Update(context.Background(), user, nil)
+		assert.True(err != nil)
+		assert.Equal("update tx is nil", err.Error())
+	})
+	t.Run("no-wrapper-WithOplog", func(t *testing.T) {
+		w := GormReadWriter{Tx: db}
+		id, err := uuid.GenerateUUID()
+		assert.Nil(err)
+		user, err := db_test.NewTestUser()
+		assert.Nil(err)
+		user.Name = "foo-" + id
+		err = w.Create(context.Background(), user)
+		assert.Nil(err)
+		assert.True(user.Id != 0)
+
+		foundUser, err := db_test.NewTestUser()
+		assert.Nil(err)
+		foundUser.PublicId = user.PublicId
+		err = w.LookupByPublicId(context.Background(), foundUser)
+		assert.Nil(err)
+		assert.Equal(foundUser.Id, user.Id)
+
+		user.FriendlyName = "friendly-" + id
+		err = w.Update(context.Background(), user, []string{"FriendlyName"},
+			WithOplog(
+				nil,
+				oplog.Metadata{
+					"key-only":   nil,
+					"deployment": []string{"amex"},
+					"project":    []string{"central-info-systems", "local-info-systems"},
+				}),
+		)
+		assert.True(err != nil)
+		assert.Equal("error wrapper is nil for WithWrapper", err.Error())
+	})
+	t.Run("no-metadata-WithOplog", func(t *testing.T) {
+		w := GormReadWriter{Tx: db}
+		id, err := uuid.GenerateUUID()
+		assert.Nil(err)
+		user, err := db_test.NewTestUser()
+		assert.Nil(err)
+		user.Name = "foo-" + id
+		err = w.Create(context.Background(), user)
+		assert.Nil(err)
+		assert.True(user.Id != 0)
+
+		foundUser, err := db_test.NewTestUser()
+		assert.Nil(err)
+		foundUser.PublicId = user.PublicId
+		err = w.LookupByPublicId(context.Background(), foundUser)
+		assert.Nil(err)
+		assert.Equal(foundUser.Id, user.Id)
+
+		user.FriendlyName = "friendly-" + id
+		err = w.Update(context.Background(), user, []string{"FriendlyName"},
+			WithOplog(
+				TestWrapper(t),
+				nil,
+			),
+		)
+		assert.True(err != nil)
+		assert.Equal("error no metadata for WithOplog", err.Error())
+	})
+}
+
+func TestGormReadWriter_Create(t *testing.T) {
+	// intentionally not run with t.Parallel so we don't need to use DoTx for the Create tests
+	cleanup, db := TestSetup(t, "migrations/postgres")
+	defer cleanup()
+	assert := assert.New(t)
+	defer db.Close()
+	t.Run("simple", func(t *testing.T) {
+		w := GormReadWriter{Tx: db}
+		id, err := uuid.GenerateUUID()
+		assert.Nil(err)
+		user, err := db_test.NewTestUser()
+		assert.Nil(err)
+		user.Name = "foo-" + id
+		err = w.Create(context.Background(), user)
+		assert.Nil(err)
+		assert.True(user.Id != 0)
+		assert.True(user.GetCreateTime() != nil)
+		assert.True(user.GetUpdateTime() != nil)
+
+		foundUser, err := db_test.NewTestUser()
+		assert.Nil(err)
+		foundUser.PublicId = user.PublicId
+		err = w.LookupByPublicId(context.Background(), foundUser)
+		assert.Nil(err)
+		assert.Equal(foundUser.Id, user.Id)
+	})
+	t.Run("valid-WithOplog", func(t *testing.T) {
+		w := GormReadWriter{Tx: db}
+		id, err := uuid.GenerateUUID()
+		assert.Nil(err)
+		user, err := db_test.NewTestUser()
+		assert.Nil(err)
+		user.Name = "foo-" + id
+		err = w.Create(
+			context.Background(),
+			user,
+			WithOplog(
+				TestWrapper(t),
+				oplog.Metadata{
+					"key-only":   nil,
+					"deployment": []string{"amex"},
+					"project":    []string{"central-info-systems", "local-info-systems"},
+				},
+			),
+		)
+		assert.Nil(err)
+		assert.True(user.Id != 0)
+
+		foundUser, err := db_test.NewTestUser()
+		assert.Nil(err)
+		foundUser.PublicId = user.PublicId
+		err = w.LookupByPublicId(context.Background(), foundUser)
+		assert.Nil(err)
+		assert.Equal(foundUser.Id, user.Id)
+	})
+	t.Run("no-wrapper-WithOplog", func(t *testing.T) {
+		w := GormReadWriter{Tx: db}
+		id, err := uuid.GenerateUUID()
+		assert.Nil(err)
+		user, err := db_test.NewTestUser()
+		assert.Nil(err)
+		user.Name = "foo-" + id
+		err = w.Create(
+			context.Background(),
+			user,
+			WithOplog(
+				nil,
+				oplog.Metadata{
+					"key-only":   nil,
+					"deployment": []string{"amex"},
+					"project":    []string{"central-info-systems", "local-info-systems"},
+				},
+			),
+		)
+		assert.True(err != nil)
+		assert.Equal("error wrapper is nil for WithWrapper", err.Error())
+	})
+	t.Run("no-metadata-WithOplog", func(t *testing.T) {
+		w := GormReadWriter{Tx: db}
+		id, err := uuid.GenerateUUID()
+		assert.Nil(err)
+		user, err := db_test.NewTestUser()
+		assert.Nil(err)
+		user.Name = "foo-" + id
+		err = w.Create(
+			context.Background(),
+			user,
+			WithOplog(
+				TestWrapper(t),
+				nil,
+			),
+		)
+		assert.True(err != nil)
+		assert.Equal("error no metadata for WithOplog", err.Error())
+	})
+	t.Run("nil-tx", func(t *testing.T) {
+		w := GormReadWriter{Tx: nil}
+		id, err := uuid.GenerateUUID()
+		assert.Nil(err)
+		user, err := db_test.NewTestUser()
+		assert.Nil(err)
+		user.Name = "foo-" + id
+		err = w.Create(context.Background(), user)
+		assert.True(err != nil)
+		assert.Equal("create tx is nil", err.Error())
+	})
+}
+
+func TestGormReadWriter_LookupByFriendlyName(t *testing.T) {
+	t.Parallel()
+	cleanup, db := TestSetup(t, "migrations/postgres")
+	defer cleanup()
+	assert := assert.New(t)
+	defer db.Close()
+	t.Run("simple", func(t *testing.T) {
+		w := GormReadWriter{Tx: db}
+		id, err := uuid.GenerateUUID()
+		assert.Nil(err)
+		user, err := db_test.NewTestUser()
+		assert.Nil(err)
+		user.Name = "foo-" + id
+		user.FriendlyName = "fn-" + id
+		err = w.Create(context.Background(), user)
+		assert.Nil(err)
+		assert.True(user.Id != 0)
+
+		foundUser, err := db_test.NewTestUser()
+		assert.Nil(err)
+		foundUser.FriendlyName = "fn-" + id
+		err = w.LookupByFriendlyName(context.Background(), foundUser)
+		assert.Nil(err)
+		assert.Equal(foundUser.Id, user.Id)
+	})
+	t.Run("tx-nil,", func(t *testing.T) {
+		w := GormReadWriter{}
+		foundUser, err := db_test.NewTestUser()
+		assert.Nil(err)
+		foundUser.FriendlyName = "fn-name"
+		err = w.LookupByFriendlyName(context.Background(), foundUser)
+		assert.True(err != nil)
+		assert.Equal("error tx nil for lookup by friendly name", err.Error())
+	})
+	t.Run("no-friendly-name-set", func(t *testing.T) {
+		w := GormReadWriter{Tx: db}
+		foundUser, err := db_test.NewTestUser()
+		assert.Nil(err)
+		err = w.LookupByFriendlyName(context.Background(), foundUser)
+		assert.True(err != nil)
+		assert.Equal("error friendly name empty string for lookup by friendly name", err.Error())
+	})
+	t.Run("not-found", func(t *testing.T) {
+		w := GormReadWriter{Tx: db}
+		id, err := uuid.GenerateUUID()
+		assert.Nil(err)
+
+		foundUser, err := db_test.NewTestUser()
+		assert.Nil(err)
+		foundUser.FriendlyName = "fn-" + id
+		err = w.LookupByFriendlyName(context.Background(), foundUser)
+		assert.True(err != nil)
+		assert.Equal(gorm.ErrRecordNotFound, err)
+	})
+}
+
+func TestGormReadWriter_LookupByPublicId(t *testing.T) {
+	t.Parallel()
+	cleanup, db := TestSetup(t, "migrations/postgres")
+	defer cleanup()
+	assert := assert.New(t)
+	defer db.Close()
+	t.Run("simple", func(t *testing.T) {
+		w := GormReadWriter{Tx: db}
+		id, err := uuid.GenerateUUID()
+		assert.Nil(err)
+		user, err := db_test.NewTestUser()
+		assert.Nil(err)
+		user.Name = "foo-" + id
+		user.FriendlyName = "fn-" + id
+		err = w.Create(context.Background(), user)
+		assert.Nil(err)
+		assert.True(user.PublicId != "")
+
+		foundUser, err := db_test.NewTestUser()
+		assert.Nil(err)
+		foundUser.PublicId = user.PublicId
+		err = w.LookupByPublicId(context.Background(), foundUser)
+		assert.Nil(err)
+		assert.Equal(foundUser.Id, user.Id)
+	})
+	t.Run("tx-nil,", func(t *testing.T) {
+		w := GormReadWriter{}
+		foundUser, err := db_test.NewTestUser()
+		assert.Nil(err)
+		err = w.LookupByPublicId(context.Background(), foundUser)
+		assert.True(err != nil)
+		assert.Equal("error tx nil for lookup by public id", err.Error())
+	})
+	t.Run("no-public-id-set", func(t *testing.T) {
+		w := GormReadWriter{Tx: db}
+		foundUser, err := db_test.NewTestUser()
+		foundUser.PublicId = ""
+		assert.Nil(err)
+		err = w.LookupByPublicId(context.Background(), foundUser)
+		assert.True(err != nil)
+		assert.Equal("error public id empty string for lookup by public id", err.Error())
+	})
+	t.Run("not-found", func(t *testing.T) {
+		w := GormReadWriter{Tx: db}
+		id, err := uuid.GenerateUUID()
+		assert.Nil(err)
+
+		foundUser, err := db_test.NewTestUser()
+		assert.Nil(err)
+		foundUser.PublicId = id
+		err = w.LookupByPublicId(context.Background(), foundUser)
+		assert.True(err != nil)
+		assert.Equal(gorm.ErrRecordNotFound, err)
+	})
+}
+
+func TestGormReadWriter_LookupWhere(t *testing.T) {
+	t.Parallel()
+	cleanup, db := TestSetup(t, "migrations/postgres")
+	defer cleanup()
+	assert := assert.New(t)
+	defer db.Close()
+	t.Run("simple", func(t *testing.T) {
+		w := GormReadWriter{Tx: db}
+		id, err := uuid.GenerateUUID()
+		assert.Nil(err)
+		user, err := db_test.NewTestUser()
+		assert.Nil(err)
+		user.Name = "foo-" + id
+		user.FriendlyName = "fn-" + id
+		err = w.Create(context.Background(), user)
+		assert.Nil(err)
+		assert.True(user.PublicId != "")
+
+		var foundUser db_test.TestUser
+		err = w.LookupWhere(context.Background(), &foundUser, "public_id = ?", user.PublicId)
+		assert.Nil(err)
+		assert.Equal(foundUser.Id, user.Id)
+	})
+	t.Run("tx-nil,", func(t *testing.T) {
+		w := GormReadWriter{}
+		var foundUser db_test.TestUser
+		err := w.LookupWhere(context.Background(), &foundUser, "public_id = ?", 1)
+		assert.True(err != nil)
+		assert.Equal("error tx nil for lookup by", err.Error())
+	})
+	t.Run("not-found", func(t *testing.T) {
+		w := GormReadWriter{Tx: db}
+		id, err := uuid.GenerateUUID()
+		assert.Nil(err)
+
+		var foundUser db_test.TestUser
+		err = w.LookupWhere(context.Background(), &foundUser, "public_id = ?", id)
+		assert.True(err != nil)
+		assert.Equal(gorm.ErrRecordNotFound, err)
+	})
+	t.Run("bad-where", func(t *testing.T) {
+		w := GormReadWriter{Tx: db}
+		id, err := uuid.GenerateUUID()
+		assert.Nil(err)
+
+		var foundUser db_test.TestUser
+		err = w.LookupWhere(context.Background(), &foundUser, "? = ?", id)
+		assert.True(err != nil)
+	})
+}
+
+func TestGormReadWriter_SearchWhere(t *testing.T) {
+	t.Parallel()
+	cleanup, db := TestSetup(t, "migrations/postgres")
+	defer cleanup()
+	assert := assert.New(t)
+	defer db.Close()
+	t.Run("simple", func(t *testing.T) {
+		w := GormReadWriter{Tx: db}
+		id, err := uuid.GenerateUUID()
+		assert.Nil(err)
+		user, err := db_test.NewTestUser()
+		assert.Nil(err)
+		user.Name = "foo-" + id
+		user.FriendlyName = "fn-" + id
+		err = w.Create(context.Background(), user)
+		assert.Nil(err)
+		assert.True(user.PublicId != "")
+
+		var foundUsers []db_test.TestUser
+		err = w.SearchWhere(context.Background(), &foundUsers, "public_id = ?", user.PublicId)
+		assert.Nil(err)
+		assert.Equal(foundUsers[0].Id, user.Id)
+	})
+	t.Run("tx-nil,", func(t *testing.T) {
+		w := GormReadWriter{}
+		var foundUsers []db_test.TestUser
+		err := w.SearchWhere(context.Background(), &foundUsers, "public_id = ?", 1)
+		assert.True(err != nil)
+		assert.Equal("error tx nil for search by", err.Error())
+	})
+	t.Run("not-found", func(t *testing.T) {
+		w := GormReadWriter{Tx: db}
+		id, err := uuid.GenerateUUID()
+		assert.Nil(err)
+
+		var foundUsers []db_test.TestUser
+		err = w.SearchWhere(context.Background(), &foundUsers, "public_id = ?", id)
+		assert.Nil(err)
+		assert.Equal(0, len(foundUsers))
+	})
+	t.Run("bad-where", func(t *testing.T) {
+		w := GormReadWriter{Tx: db}
+		id, err := uuid.GenerateUUID()
+		assert.Nil(err)
+
+		var foundUsers []db_test.TestUser
+		err = w.SearchWhere(context.Background(), &foundUsers, "? = ?", id)
+		assert.True(err != nil)
+	})
+}
+
+func TestGormReadWriter_DB(t *testing.T) {
+	t.Parallel()
+	cleanup, db := TestSetup(t, "migrations/postgres")
+	defer cleanup()
+	assert := assert.New(t)
+	defer db.Close()
+	t.Run("valid", func(t *testing.T) {
+		w := GormReadWriter{Tx: db}
+		d, err := w.DB()
+		assert.Nil(err)
+		assert.True(d != nil)
+		err = d.Ping()
+		assert.Nil(err)
+	})
+	t.Run("nil-tx", func(t *testing.T) {
+		w := GormReadWriter{Tx: nil}
+		d, err := w.DB()
+		assert.True(err != nil)
+		assert.True(d == nil)
+		assert.Equal("create tx is nil for db", err.Error())
+	})
+}
+
+func TestGormReadWriter_DoTx(t *testing.T) {
+	t.Parallel()
+	cleanup, db := TestSetup(t, "migrations/postgres")
+	defer cleanup()
+	assert := assert.New(t)
+	defer db.Close()
+
+	t.Run("valid-with-10-retries", func(t *testing.T) {
+		w := &GormReadWriter{Tx: db}
+		attempts := 0
+		got, err := w.DoTx(context.Background(), 10, ExpBackoff{},
+			func(Writer) error {
+				attempts += 1
+				if attempts < 9 {
+					return oplog.ErrTicketAlreadyRedeemed
+				}
+				return nil
+			})
+		assert.Nil(err)
+		assert.Equal(8, got.Retries)
+		assert.Equal(9, attempts) // attempted 1 + 8 retries
+	})
+	t.Run("valid-with-1-retries", func(t *testing.T) {
+		w := &GormReadWriter{Tx: db}
+		attempts := 0
+		got, err := w.DoTx(context.Background(), 1, ExpBackoff{},
+			func(Writer) error {
+				attempts += 1
+				if attempts < 2 {
+					return oplog.ErrTicketAlreadyRedeemed
+				}
+				return nil
+			})
+		assert.Nil(err)
+		assert.Equal(1, got.Retries)
+		assert.Equal(2, attempts) // attempted 1 + 8 retries
+	})
+	t.Run("valid-with-2-retries", func(t *testing.T) {
+		w := &GormReadWriter{Tx: db}
+		attempts := 0
+		got, err := w.DoTx(context.Background(), 3, ExpBackoff{},
+			func(Writer) error {
+				attempts += 1
+				if attempts < 3 {
+					return oplog.ErrTicketAlreadyRedeemed
+				}
+				return nil
+			})
+		assert.Nil(err)
+		assert.Equal(2, got.Retries)
+		assert.Equal(3, attempts) // attempted 1 + 8 retries
+	})
+	t.Run("valid-with-4-retries", func(t *testing.T) {
+		w := &GormReadWriter{Tx: db}
+		attempts := 0
+		got, err := w.DoTx(context.Background(), 4, ExpBackoff{},
+			func(Writer) error {
+				attempts += 1
+				if attempts < 4 {
+					return oplog.ErrTicketAlreadyRedeemed
+				}
+				return nil
+			})
+		assert.Nil(err)
+		assert.Equal(3, got.Retries)
+		assert.Equal(4, attempts) // attempted 1 + 8 retries
+	})
+	t.Run("zero-retries", func(t *testing.T) {
+		w := &GormReadWriter{Tx: db}
+		attempts := 0
+		got, err := w.DoTx(context.Background(), 0, ExpBackoff{}, func(Writer) error { attempts += 1; return nil })
+		assert.Nil(err)
+		assert.Equal(RetryInfo{}, got)
+		assert.Equal(1, attempts)
+	})
+	t.Run("nil-tx", func(t *testing.T) {
+		w := &GormReadWriter{nil}
+		attempts := 0
+		got, err := w.DoTx(context.Background(), 1, ExpBackoff{}, func(Writer) error { attempts += 1; return nil })
+		assert.True(err != nil)
+		assert.Equal(RetryInfo{}, got)
+		assert.Equal("do tx is nil", err.Error())
+	})
+	t.Run("not-a-retry-err", func(t *testing.T) {
+		w := &GormReadWriter{Tx: db}
+		got, err := w.DoTx(context.Background(), 1, ExpBackoff{}, func(Writer) error { return errors.New("not a retry error") })
+		assert.True(err != nil)
+		assert.Equal(RetryInfo{}, got)
+		assert.True(err != oplog.ErrTicketAlreadyRedeemed)
+	})
+	t.Run("too-many-retries", func(t *testing.T) {
+		w := &GormReadWriter{Tx: db}
+		attempts := 0
+		got, err := w.DoTx(context.Background(), 2, ExpBackoff{}, func(Writer) error { attempts += 1; return oplog.ErrTicketAlreadyRedeemed })
+		assert.True(err != nil)
+		assert.Equal(3, got.Retries)
+		assert.Equal("Too many retries: 3 of 3", err.Error())
+	})
+}
+
+func TestGormReadWriter_Delete(t *testing.T) {
+	// intentionally not run with t.Parallel so we don't need to use DoTx for the Create tests
+	cleanup, db := TestSetup(t, "migrations/postgres")
+	defer cleanup()
+	assert := assert.New(t)
+	defer db.Close()
+	t.Run("simple", func(t *testing.T) {
+		w := GormReadWriter{Tx: db}
+		id, err := uuid.GenerateUUID()
+		assert.Nil(err)
+		user, err := db_test.NewTestUser()
+		assert.Nil(err)
+		user.Name = "foo-" + id
+		err = w.Create(context.Background(), user)
+		assert.Nil(err)
+		assert.True(user.Id != 0)
+		assert.True(user.GetCreateTime() != nil)
+		assert.True(user.GetUpdateTime() != nil)
+
+		foundUser, err := db_test.NewTestUser()
+		assert.Nil(err)
+		foundUser.PublicId = user.PublicId
+		err = w.LookupByPublicId(context.Background(), foundUser)
+		assert.Nil(err)
+		assert.Equal(foundUser.Id, user.Id)
+
+		err = w.Delete(context.Background(), user)
+		assert.Nil(err)
+
+		err = w.LookupByPublicId(context.Background(), foundUser)
+		assert.True(err != nil)
+		assert.Equal(gorm.ErrRecordNotFound, err)
+	})
+	t.Run("valid-WithOplog", func(t *testing.T) {
+		w := GormReadWriter{Tx: db}
+		id, err := uuid.GenerateUUID()
+		assert.Nil(err)
+		user, err := db_test.NewTestUser()
+		assert.Nil(err)
+		user.Name = "foo-" + id
+		err = w.Create(
+			context.Background(),
+			user,
+			WithOplog(
+				TestWrapper(t),
+				oplog.Metadata{
+					"key-only":   nil,
+					"deployment": []string{"amex"},
+					"project":    []string{"central-info-systems", "local-info-systems"},
+				},
+			),
+		)
+		assert.Nil(err)
+		assert.True(user.Id != 0)
+
+		foundUser, err := db_test.NewTestUser()
+		assert.Nil(err)
+		foundUser.PublicId = user.PublicId
+		err = w.LookupByPublicId(context.Background(), foundUser)
+		assert.Nil(err)
+		assert.Equal(foundUser.Id, user.Id)
+
+		err = w.Delete(
+			context.Background(),
+			user,
+			WithOplog(
+				TestWrapper(t),
+				oplog.Metadata{
+					"key-only":   nil,
+					"deployment": []string{"amex"},
+					"project":    []string{"central-info-systems", "local-info-systems"},
+				},
+			),
+		)
+		assert.Nil(err)
+
+		err = w.LookupByPublicId(context.Background(), foundUser)
+		assert.True(err != nil)
+		assert.Equal(gorm.ErrRecordNotFound, err)
+	})
+	t.Run("no-wrapper-WithOplog", func(t *testing.T) {
+		w := GormReadWriter{Tx: db}
+		id, err := uuid.GenerateUUID()
+		assert.Nil(err)
+		user, err := db_test.NewTestUser()
+		assert.Nil(err)
+		user.Name = "foo-" + id
+		err = w.Create(
+			context.Background(),
+			user,
+			WithOplog(
+				TestWrapper(t),
+				oplog.Metadata{
+					"key-only":   nil,
+					"deployment": []string{"amex"},
+					"project":    []string{"central-info-systems", "local-info-systems"},
+				},
+			),
+		)
+		assert.Nil(err)
+		assert.True(user.Id != 0)
+
+		foundUser, err := db_test.NewTestUser()
+		assert.Nil(err)
+		foundUser.PublicId = user.PublicId
+		err = w.LookupByPublicId(context.Background(), foundUser)
+		assert.Nil(err)
+		assert.Equal(foundUser.Id, user.Id)
+
+		err = w.Delete(
+			context.Background(),
+			user,
+			WithOplog(
+				nil,
+				oplog.Metadata{
+					"key-only":   nil,
+					"deployment": []string{"amex"},
+					"project":    []string{"central-info-systems", "local-info-systems"},
+				},
+			),
+		)
+		assert.True(err != nil)
+		assert.Equal("error wrapper is nil for WithWrapper", err.Error())
+	})
+	t.Run("no-metadata-WithOplog", func(t *testing.T) {
+		w := GormReadWriter{Tx: db}
+		id, err := uuid.GenerateUUID()
+		assert.Nil(err)
+		user, err := db_test.NewTestUser()
+		assert.Nil(err)
+		user.Name = "foo-" + id
+		err = w.Create(
+			context.Background(),
+			user,
+			WithOplog(
+				TestWrapper(t),
+				oplog.Metadata{
+					"key-only":   nil,
+					"deployment": []string{"amex"},
+					"project":    []string{"central-info-systems", "local-info-systems"},
+				},
+			),
+		)
+		assert.Nil(err)
+		assert.True(user.Id != 0)
+
+		foundUser, err := db_test.NewTestUser()
+		assert.Nil(err)
+		foundUser.PublicId = user.PublicId
+		err = w.LookupByPublicId(context.Background(), foundUser)
+		assert.Nil(err)
+		assert.Equal(foundUser.Id, user.Id)
+
+		err = w.Delete(
+			context.Background(),
+			user,
+			WithOplog(
+				TestWrapper(t),
+				nil,
+			),
+		)
+		assert.True(err != nil)
+		assert.Equal("error no metadata for WithOplog", err.Error())
+	})
+	t.Run("nil-tx", func(t *testing.T) {
+		w := GormReadWriter{Tx: nil}
+		id, err := uuid.GenerateUUID()
+		assert.Nil(err)
+		user, err := db_test.NewTestUser()
+		assert.Nil(err)
+		user.Name = "foo-" + id
+		err = w.Create(context.Background(), user)
+		assert.True(err != nil)
+		assert.Equal("create tx is nil", err.Error())
+	})
+}
+
+func TestGormReadWriter_ScanRows(t *testing.T) {
+	t.Parallel()
+	cleanup, db := TestSetup(t, "migrations/postgres")
+	defer cleanup()
+	assert := assert.New(t)
+	defer db.Close()
+	t.Run("valid", func(t *testing.T) {
+		w := GormReadWriter{Tx: db}
+		user, err := db_test.NewTestUser()
+		assert.Nil(err)
+		err = w.Create(context.Background(), user)
+		assert.Nil(err)
+		assert.True(user.Id != 0)
+
+		tx, err := w.DB()
+		where := "select * from db_test_user where friendly_name in ($1, $2)"
+		rows, err := tx.Query(where, "alice", "bob")
+		defer rows.Close()
+		for rows.Next() {
+			u, err := db_test.NewTestUser()
+			assert.Nil(err)
+
+			// scan the row into your Gorm struct
+			err = w.ScanRows(rows, &u)
+			assert.Nil(err)
+			assert.Equal(user.PublicId, u.PublicId)
+		}
+	})
+}

internal/db/testing.go

@@ -0,0 +1,125 @@
+package db
+
+import (
+	"crypto/rand"
+	"database/sql"
+	"fmt"
+	"os"
+	"strings"
+	"testing"
+
+	"github.com/golang-migrate/migrate/v4"
+	_ "github.com/golang-migrate/migrate/v4/database/postgres"
+	_ "github.com/golang-migrate/migrate/v4/source/file"
+	wrapping "github.com/hashicorp/go-kms-wrapping"
+	"github.com/hashicorp/go-kms-wrapping/wrappers/aead"
+	"github.com/jinzhu/gorm"
+	"github.com/ory/dockertest/v3"
+)
+
+// setup the tests (initialize the database one-time and intialized testDatabaseURL)
+func TestSetup(t *testing.T, migrationsDirectory string) (func(), *gorm.DB) {
+	if _, err := os.Stat(migrationsDirectory); os.IsNotExist(err) {
+		t.Fatal("error migrationsDirectory does not exist")
+	}
+
+	cleanup := func() {}
+	var url string
+	var err error
+	cleanup, url, err = initDbInDocker(t, migrationsDirectory)
+	if err != nil {
+		t.Fatal(err)
+	}
+	db, err := gorm.Open("postgres", url)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return cleanup, db
+}
+
+// TestWrapper initializes an AEAD wrapping.Wrapper for testing the oplog
+func TestWrapper(t *testing.T) wrapping.Wrapper {
+	rootKey := make([]byte, 32)
+	n, err := rand.Read(rootKey)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if n != 32 {
+		t.Fatal(n)
+	}
+	root := aead.NewWrapper(nil)
+	if err := root.SetAESGCMKeyBytes(rootKey); err != nil {
+		t.Fatal(err)
+	}
+	return root
+}
+
+// initDbInDocker initializes postgres within dockertest for the unit tests
+func initDbInDocker(t *testing.T, migrationsDirectory string) (cleanup func(), retURL string, err error) {
+	if os.Getenv("PG_URL") != "" {
+		TestInitStore(t, func() {}, os.Getenv("PG_URL"), migrationsDirectory)
+		return func() {}, os.Getenv("PG_URL"), nil
+	}
+	pool, err := dockertest.NewPool("")
+	if err != nil {
+		return func() {}, "", fmt.Errorf("could not connect to docker: %w", err)
+	}
+
+	resource, err := pool.Run("postgres", "latest", []string{"POSTGRES_PASSWORD=secret", "POSTGRES_DB=watchtower"})
+	if err != nil {
+		return func() {}, "", fmt.Errorf("could not start resource: %w", err)
+	}
+
+	c := func() {
+		cleanupResource(t, pool, resource)
+	}
+
+	url := fmt.Sprintf("postgres://postgres:secret@localhost:%s?sslmode=disable", resource.GetPort("5432/tcp"))
+
+	if err := pool.Retry(func() error {
+		db, err := sql.Open("postgres", url)
+		if err != nil {
+			return fmt.Errorf("error opening postgres dev container: %w", err)
+		}
+
+		if err := db.Ping(); err != nil {
+			return err
+		}
+		defer db.Close()
+		return nil
+	}); err != nil {
+		return func() {}, "", fmt.Errorf("could not connect to docker: %w", err)
+	}
+	TestInitStore(t, c, url, migrationsDirectory)
+	return c, url, nil
+}
+
+// TestInitStore will execute the migrations needed to initialize the store for tests
+func TestInitStore(t *testing.T, cleanup func(), url string, migrationsDirectory string) {
+	// run migrations
+	m, err := migrate.New(fmt.Sprintf("file://%s", migrationsDirectory), url)
+	if err != nil {
+		cleanup()
+		t.Fatalf("Error creating migrations: %s", err)
+	}
+	if err := m.Up(); err != nil && err != migrate.ErrNoChange {
+		cleanup()
+		t.Fatalf("Error running migrations: %s", err)
+	}
+}
+
+// cleanupResource will clean up the dockertest resources (postgres)
+func cleanupResource(t *testing.T, pool *dockertest.Pool, resource *dockertest.Resource) {
+	var err error
+	for i := 0; i < 10; i++ {
+		err = pool.Purge(resource)
+		if err == nil {
+			return
+		}
+	}
+
+	if strings.Contains(err.Error(), "No such container") {
+		return
+	}
+	t.Fatalf("Failed to cleanup local container: %s", err)
+}

internal/db/testing_test.go

@@ -0,0 +1,15 @@
+package db
+
+import (
+	"testing"
+)
+
+func Test_Utils(t *testing.T) {
+	t.Parallel()
+	cleanup, conn := TestSetup(t, "migrations/postgres")
+	defer cleanup()
+	defer conn.Close()
+	t.Run("nothing", func(t *testing.T) {
+
+	})
+}

internal/proto/local/controller/storage/db/db_test/v1/db_test.proto

@@ -0,0 +1,88 @@
+syntax = "proto3";
+
+// define a test proto package for the internal/db package.  These protos
+// are only used for unit tests and are not part of the rest of the watchtower
+// domain model
+package controller.storage.db.db_test.v1;
+option go_package = "github.com/hashicorp/watchtower/internal/db/db_test;db_test";
+
+import "google/protobuf/timestamp.proto";
+
+// Timestamp for oplog messages
+message Timestamp {
+  google.protobuf.Timestamp timestamp = 1;
+}
+
+// TestUser for gorm test user model
+message StoreTestUser {
+  // @inject_tag: gorm:"primary_key"
+  uint32 id = 1;
+
+  // create_time from the RDBMS
+  // @inject_tag: `gorm:"default:CURRENT_TIMESTAMP"`
+  Timestamp create_time = 2;
+
+  // update_time from the RDBMS
+  // @inject_tag: `gorm:"default:CURRENT_TIMESTAMP"`
+  Timestamp update_time = 3;
+
+  // public_id is the used to access the AssignablScope via an API
+  string public_id = 4;
+
+  // FriendlyName is the optional friendly name used to
+  // access the Scope via an API
+  // @inject_tag: `gorm:"default:null"`
+  string friendly_name = 5;
+
+  string name = 6;
+  string phone_number = 7;
+  string email = 8;
+}
+// TestCar for gorm test car model
+message StoreTestCar {
+  // @inject_tag: gorm:"primary_key"
+  uint32 id = 1;
+
+  // create_time from the RDBMS
+  // @inject_tag: `gorm:"default:CURRENT_TIMESTAMP"`
+  Timestamp create_time = 2;
+
+  // update_time from the RDBMS
+  // @inject_tag: `gorm:"default:CURRENT_TIMESTAMP"`
+  Timestamp update_time = 3;
+
+  // public_id is the used to access the AssignablScope via an API
+  string public_id = 4;
+
+  // FriendlyName is the optional friendly name used to
+  // access the Scope via an API
+  // @inject_tag: `gorm:"default:null"`
+  string friendly_name = 5;
+
+  string model = 6;
+  int32 mpg = 7;
+}
+// TestRental for gorm test rental model
+message StoreTestRental {
+  // @inject_tag: gorm:"primary_key"
+  uint32 id = 1;
+
+  // create_time from the RDBMS
+  // @inject_tag: `gorm:"default:CURRENT_TIMESTAMP"`
+  Timestamp create_time = 2;
+
+  // update_time from the RDBMS
+  // @inject_tag: `gorm:"default:CURRENT_TIMESTAMP"`
+  Timestamp update_time = 3;
+
+  // public_id is the used to access the AssignablScope via an API
+  string public_id = 4;
+
+  // FriendlyName is the optional friendly name used to
+  // access the Scope via an API
+  // @inject_tag: `gorm:"default:null"`
+  string friendly_name = 5;
+
+  uint32 user_id = 6;
+  uint32 car_id = 7;
+}