work

README

work

A compact library for tracking and committing atomic changes to your entities.

Demo

make demo (requires docker-compose).

How to use it?

Construction

Starting with entities Foo and Bar,

// entities.
f, b := Foo{}, Bar{}

// type names.
ft, bt := unit.TypeNameOf(f), unit.TypeNameOf(b)

// data mappers.
m := map[unit.TypeName]unit.DataMapper { ft: fdm, bt: fdm }

// 🎉
opts = []unit.Option{ unit.DB(db), unit.DataMappers(m) }
unit, err := unit.New(opts...)

Adding

When creating new entities, use Add:

additions := []interface{}{ f, b }
err := u.Add(additions...)

Updating

When modifying existing entities, use Alter:

updates := []interface{}{ f, b }
err := u.Alter(updates...)

Removing

When removing existing entities, use Remove:

removals := []interface{}{ f, b }
err := u.Remove(removals...)

Registering

When retrieving existing entities, track their intial state using Register:

fetched := []interface{}{ f, b }
err := u.Register(fetched...)

Saving

When you are ready to commit your work unit, use Save:

ctx := context.Background()
err := u.Save(ctx)

Logging

We use zap as our logging library of choice. To leverage the logs emitted from the work units, utilize the unit.Logger option with an instance of *zap.Logger upon creation:

// create logger.
l, _ := zap.NewDevelopment()

opts = []unit.Option{
	unit.DB(db),
	unit.DataMappers(m),
	unit.Logger(l), // 🎉
}
u, err := unit.New(opts...)

Metrics

For emitting metrics, we use tally. To utilize the metrics emitted from the work units, leverage the unit.Scope option with a tally.Scope upon creation. Assuming we have a scope s, it would look like so:

opts = []unit.Option{
	unit.DB(db),
	unit.DataMappers(m),
	unit.Scope(s), // 🎉
}
u, err := unit.New(opts...)

Emitted Metrics

Name	Type	Description
[PREFIX.]unit.save.success	counter	The number of successful work unit saves.
[PREFIX.]unit.save	timer	The time duration when saving a work unit.
[PREFIX.]unit.rollback.success	counter	The number of successful work unit rollbacks.
[PREFIX.]unit.rollback.failure	counter	The number of unsuccessful work unit rollbacks.
[PREFIX.]unit.rollback	timer	The time duration when rolling back a work unit.
[PREFIX.]unit.retry.attempt	counter	The number of retry attempts.
[PREFIX.]unit.insert	counter	The number of successful inserts performed.
[PREFIX.]unit.update	counter	The number of successful updates performed.
[PREFIX.]unit.delete	counter	The number of successful deletes performed.

Uniters

In most circumstances, an application has many aspects that result in the creation of a work unit. To tackle that challenge, we recommend using unit.Uniter to create instances of unit., like so:

opts = []unit.Option{
	unit.DB(db),
	unit.DataMappers(m),
	unit.Logger(l),
}
uniter := unit.NewUniter(opts...)

// create the unit.
u, err := uniter.Unit()

Documentation

Overview

Copyright 2021 Freerware * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License.

Index

• Constants
• Variables
• type DataMapper
• type MapperContext
• type TypeName
  • func TypeNameOf(entity interface{}) TypeName
  • func (t TypeName) String() string
• type Unit
  • func NewUnit(opts ...UnitOption) (Unit, error)
• type UnitAction
• type UnitActionContext
• type UnitActionType
• type UnitOption
• type UnitOptions
• type UnitRetryDelayType
• type Uniter
  • func NewUniter(options ...UnitOption) Uniter

Constants

const (
	// UnitActionTypeAfterRegister indicates an action type that occurs after an entity is registered.
	UnitActionTypeAfterRegister = iota
	// UnitActionTypeAfterAdd indicates an action type that occurs after an entity is added.
	UnitActionTypeAfterAdd
	// UnitActionTypeAfterAlter indicates an action type that occurs after an entity is altered.
	UnitActionTypeAfterAlter
	// UnitActionTypeAfterRemove indicates an action type that occurs after an entity is removed.
	UnitActionTypeAfterRemove
	// UnitActionTypeAfterInserts indicates an action type that occurs after new entities are inserted in the data store.
	UnitActionTypeAfterInserts
	// UnitActionTypeAfterUpdates indicates an action type that occurs after existing entities are updated in the data store.
	UnitActionTypeAfterUpdates
	// UnitActionTypeAfterDeletes indicates an action type that occurs after existing entities are deleted in the data store.
	UnitActionTypeAfterDeletes
	// UnitActionTypeAfterRollback indicates an action type that occurs after rollback.
	UnitActionTypeAfterRollback
	// UnitActionTypeAfterSave indicates an action type that occurs after save.
	UnitActionTypeAfterSave
	// UnitActionTypeBeforeRegister indicates an action type that occurs before an entity is registered.
	UnitActionTypeBeforeRegister
	// UnitActionTypeBeforeAdd indicates an action type that occurs before an entity is added.
	UnitActionTypeBeforeAdd
	// UnitActionTypeBeforeAlter indicates an action type that occurs before an entity is altered.
	UnitActionTypeBeforeAlter
	// UnitActionTypeBeforeRemove indicates an action type that occurs before an entity is removed.
	UnitActionTypeBeforeRemove
	// UnitActionTypeBeforeInserts indicates an action type that occurs before new entities are inserted in the data store.
	UnitActionTypeBeforeInserts
	// UnitActionTypeBeforeUpdates indicates an action type that occurs before existing entities are updated in the data store.
	UnitActionTypeBeforeUpdates
	// UnitActionTypeBeforeDeletes indicates an action type that occurs before existing entities are deleted in the data store.
	UnitActionTypeBeforeDeletes
	// UnitActionTypeBeforeRollback indicates an action type that occurs before rollback.
	UnitActionTypeBeforeRollback
	// UnitActionTypeBeforeSave indicates an action type that occurs before save.
	UnitActionTypeBeforeSave
)

The various types of actions that are executed throughout the lifecycle of a work unit.

const (
	// Fixed represents a retry type that maintains a constaint delay between retry iterations.
	UnitRetryDelayTypeFixed = iota
	// BackOff represents a retry type that increases delay between retry iterations.
	UnitRetryDelayTypeBackOff
	// Random represents a retry type that utilizes a random delay between retry iterations.
	UnitRetryDelayTypeRandom
)

Variables

var (

	// ErrMissingDataMapper represents the error that is returned
	// when attempting to add, alter, remove, or register an entity
	// that doesn't have a corresponding data mapper.
	ErrMissingDataMapper = errors.New("missing data mapper for entity")

	// ErrNoDataMapper represents the error that occurs when attempting
	// to create a work unit without any data mappers.
	ErrNoDataMapper = errors.New("must have at least one data mapper")
)

var (
	// UnitDB specifies the option to provide the database for the work unit.
	UnitDB = func(db *sql.DB) UnitOption {
		return func(o *UnitOptions) {
			o.DB = db
		}
	}

	// UnitDataMappers specifies the option to provide the data mappers for the work unit.
	UnitDataMappers = func(dm map[TypeName]DataMapper) UnitOption {
		return func(o *UnitOptions) {
			if o.DataMappers == nil {
				o.DataMappers = make(map[TypeName]DataMapper)
			}
			o.DataMappers = dm
		}
	}

	// UnitLogger specifies the option to provide a logger for the work unit.
	UnitLogger = func(l *zap.Logger) UnitOption {
		return func(o *UnitOptions) {
			o.Logger = l
		}
	}

	// UnitScope specifies the option to provide a metric scope for the work unit.
	UnitScope = func(s tally.Scope) UnitOption {
		return func(o *UnitOptions) {
			o.Scope = s
		}
	}

	// UnitAfterRegisterActions specifies the option to provide actions to execute
	// after entities are registered with the work unit.
	UnitAfterRegisterActions = func(a ...UnitAction) UnitOption {
		return setActions(UnitActionTypeAfterRegister, a...)
	}

	// UnitAfterAddActions specifies the option to provide actions to execute
	// after entities are added with the work unit.
	UnitAfterAddActions = func(a ...UnitAction) UnitOption {
		return setActions(UnitActionTypeAfterAdd, a...)
	}

	// UnitAfterAlterActions specifies the option to provide actions to execute
	// after entities are altered with the work unit.
	UnitAfterAlterActions = func(a ...UnitAction) UnitOption {
		return setActions(UnitActionTypeAfterAlter, a...)
	}

	// UnitAfterRemoveActions specifies the option to provide actions to execute
	// after entities are removed with the work unit.
	UnitAfterRemoveActions = func(a ...UnitAction) UnitOption {
		return setActions(UnitActionTypeAfterRemove, a...)
	}

	// UnitAfterInsertsActions specifies the option to provide actions to execute
	// after new entities are inserted in the data store.
	UnitAfterInsertsActions = func(a ...UnitAction) UnitOption {
		return setActions(UnitActionTypeAfterInserts, a...)
	}

	// UnitAfterUpdatesActions specifies the option to provide actions to execute
	// after altered entities are updated in the data store.
	UnitAfterUpdatesActions = func(a ...UnitAction) UnitOption {
		return setActions(UnitActionTypeAfterUpdates, a...)
	}

	// UnitAfterDeletesActions specifies the option to provide actions to execute
	// after removed entities are deleted in the data store.
	UnitAfterDeletesActions = func(a ...UnitAction) UnitOption {
		return setActions(UnitActionTypeAfterDeletes, a...)
	}

	// UnitAfterRollbackActions specifies the option to provide actions to execute
	// after a rollback is performed.
	UnitAfterRollbackActions = func(a ...UnitAction) UnitOption {
		return setActions(UnitActionTypeAfterRollback, a...)
	}

	// UnitAfterSaveActions specifies the option to provide actions to execute
	// after a save is performed.
	UnitAfterSaveActions = func(a ...UnitAction) UnitOption {
		return setActions(UnitActionTypeAfterSave, a...)
	}

	// UnitBeforeInsertsActions specifies the option to provide actions to execute
	// before new entities are inserted in the data store.
	UnitBeforeInsertsActions = func(a ...UnitAction) UnitOption {
		return setActions(UnitActionTypeBeforeInserts, a...)
	}

	// UnitBeforeUpdatesActions specifies the option to provide actions to execute
	// before altered entities are updated in the data store.
	UnitBeforeUpdatesActions = func(a ...UnitAction) UnitOption {
		return setActions(UnitActionTypeBeforeUpdates, a...)
	}

	// UnitBeforeDeletesActions specifies the option to provide actions to execute
	// before removed entities are deleted in the data store.
	UnitBeforeDeletesActions = func(a ...UnitAction) UnitOption {
		return setActions(UnitActionTypeBeforeDeletes, a...)
	}

	// UnitBeforeRollbackActions specifies the option to provide actions to execute
	// before a rollback is performed.
	UnitBeforeRollbackActions = func(a ...UnitAction) UnitOption {
		return setActions(UnitActionTypeBeforeRollback, a...)
	}

	// UnitBeforeSaveActions specifies the option to provide actions to execute
	// before a save is performed.
	UnitBeforeSaveActions = func(a ...UnitAction) UnitOption {
		return setActions(UnitActionTypeBeforeSave, a...)
	}

	// UnitDefaultLoggingActions specifies all of the default logging actions.
	UnitDefaultLoggingActions = func() UnitOption {
		beforeInsertLogAction := func(ctx UnitActionContext) {
			ctx.Logger.Debug(
				"attempting to insert entities",
				zap.Int("count", ctx.AdditionCount),
			)
		}
		afterInsertLogAction := func(ctx UnitActionContext) {
			ctx.Logger.Debug(
				"successfully inserted entities",
				zap.Int("count", ctx.AdditionCount),
			)
		}
		beforeUpdateLogAction := func(ctx UnitActionContext) {
			ctx.Logger.Debug(
				"attempting to update entities",
				zap.Int("count", ctx.AlterationCount),
			)
		}
		afterUpdateLogAction := func(ctx UnitActionContext) {
			ctx.Logger.Debug(
				"successfully updated entities",
				zap.Int("count", ctx.AlterationCount),
			)
		}
		beforeDeleteLogAction := func(ctx UnitActionContext) {
			ctx.Logger.Debug(
				"attempting to delete entities",
				zap.Int("count", ctx.RemovalCount),
			)
		}
		afterDeleteLogAction := func(ctx UnitActionContext) {
			ctx.Logger.Debug(
				"successfully deleted entities",
				zap.Int("count", ctx.RemovalCount),
			)
		}
		beforeSaveLogAction := func(ctx UnitActionContext) {
			ctx.Logger.Debug("attempting to save unit")
		}
		afterSaveLogAction := func(ctx UnitActionContext) {
			totalCount :=
				ctx.AdditionCount + ctx.AlterationCount + ctx.RemovalCount
			ctx.Logger.Info("successfully saved unit",
				zap.Int("insertCount", ctx.AdditionCount),
				zap.Int("updateCount", ctx.AlterationCount),
				zap.Int("deleteCount", ctx.RemovalCount),
				zap.Int("registerCount", ctx.RegisterCount),
				zap.Int("totalUpdateCount", totalCount))
		}
		beforeRollbackLogAction := func(ctx UnitActionContext) {
			ctx.Logger.Debug("attempting to roll back unit")
		}
		afterRollbackLogAction := func(ctx UnitActionContext) {
			ctx.Logger.Info("successfully rolled back unit")
		}
		return func(o *UnitOptions) {
			subOpts := []UnitOption{
				setActions(UnitActionTypeBeforeInserts, beforeInsertLogAction),
				setActions(UnitActionTypeAfterInserts, afterInsertLogAction),
				setActions(UnitActionTypeBeforeUpdates, beforeUpdateLogAction),
				setActions(UnitActionTypeAfterUpdates, afterUpdateLogAction),
				setActions(UnitActionTypeBeforeDeletes, beforeDeleteLogAction),
				setActions(UnitActionTypeAfterDeletes, afterDeleteLogAction),
				setActions(UnitActionTypeBeforeSave, beforeSaveLogAction),
				setActions(UnitActionTypeAfterSave, afterSaveLogAction),
				setActions(UnitActionTypeBeforeRollback, beforeRollbackLogAction),
				setActions(UnitActionTypeAfterRollback, afterRollbackLogAction),
			}
			for _, opt := range subOpts {
				opt(o)
			}
		}
	}

	// DisableDefaultLoggingActions disables the default logging actions.
	DisableDefaultLoggingActions = func() UnitOption {
		return func(o *UnitOptions) {
			o.DisableDefaultLoggingActions = true
		}
	}

	// UnitRetryAttempts defines the number of retry attempts to perform.
	UnitRetryAttempts = func(attempts int) UnitOption {
		if attempts < 0 {
			attempts = 0
		}
		return func(o *UnitOptions) {
			o.RetryAttempts = attempts
		}
	}

	// UnitRetryDelay defines the delay to utilize during retries.
	UnitRetryDelay = func(delay time.Duration) UnitOption {
		return func(o *UnitOptions) {
			o.RetryDelay = delay
		}
	}

	// UnitRetryMaximumJitter defines the maximum jitter to utilize during
	// retries that utilize random delay times.
	UnitRetryMaximumJitter = func(jitter time.Duration) UnitOption {
		return func(o *UnitOptions) {
			o.RetryMaximumJitter = jitter
		}
	}

	// UnitRetryType defines the type of retry to perform.
	UnitRetryType = func(retryType UnitRetryDelayType) UnitOption {
		return func(o *UnitOptions) {
			o.RetryType = retryType
		}
	}
)

Types

type DataMapper

type DataMapper interface {
	Insert(context.Context, MapperContext, ...interface{}) error
	Update(context.Context, MapperContext, ...interface{}) error
	Delete(context.Context, MapperContext, ...interface{}) error
}

DataMapper represents a creator, modifier, and deleter of entities.

type MapperContext

type MapperContext struct {
	Tx *sql.Tx
}

type TypeName

type TypeName string

TypeName represents an entity's type.

func TypeNameOf

func TypeNameOf(entity interface{}) TypeName

TypeNameOf provides the type name for the provided entity.

func (TypeName) String

func (t TypeName) String() string

String provides the string representation of the type name.

type Unit

type Unit interface {

	// Register tracks the provided entities as clean.
	Register(...interface{}) error

	// Add marks the provided entities as new additions.
	Add(...interface{}) error

	// Alter marks the provided entities as modifications.
	Alter(...interface{}) error

	// Remove marks the provided entities as removals.
	Remove(...interface{}) error

	// Save commits the new additions, modifications, and removals
	// within the work unit to a persistent store.
	Save(context.Context) error
}

Unit represents an atomic set of entity changes.

func NewUnit

func NewUnit(opts ...UnitOption) (Unit, error)

type UnitAction

type UnitAction func(UnitActionContext)

Action represents an operation performed during a paticular lifecycle event of a work unit.

type UnitActionContext

type UnitActionContext struct {
	// Logger is the work units configured logger.
	Logger *zap.Logger
	// Scope is the work units configured metrics scope.
	Scope tally.Scope
	// AdditionCount represents the number of entities indicated as new.
	AdditionCount int
	// AlterationCount represents the number of entities indicated as modified.
	AlterationCount int
	// RemovalCount represents the number of entities indicated as removed.
	RemovalCount int
	// RegisterCount represents the number of entities indicated as registered.
	RegisterCount int
}

UnitActionContext represents the executional context for an action.

type UnitActionType

type UnitActionType int

UnitActionType represents the type of work unit action.

type UnitOption

type UnitOption func(*UnitOptions)

UnitOption applies an option to the provided configuration.

type UnitOptions

type UnitOptions struct {
	Logger                       *zap.Logger
	Scope                        tally.Scope
	Actions                      map[UnitActionType][]UnitAction
	DisableDefaultLoggingActions bool
	DataMappers                  map[TypeName]DataMapper
	DB                           *sql.DB
	RetryAttempts                int
	RetryDelay                   time.Duration
	RetryMaximumJitter           time.Duration
	RetryType                    UnitRetryDelayType
}

UnitOptions represents the configuration options for the work unit.

type UnitRetryDelayType

type UnitRetryDelayType int

UnitRetryDelayType represents the type of retry delay to perform.

type Uniter

type Uniter interface {

	//Unit constructs a new work unit.
	Unit() (Unit, error)
}

Uniter represents a factory for work units.

func NewUniter

func NewUniter(options ...UnitOption) Uniter

NewUniter creates a new uniter with the provided unit options.