db

Documentation

Index

• Constants
• Variables
• type Database
• type DynamoDb
  • func NewDynamoDb(cfg aws.Config, tableName string) *DynamoDb
  • func (db *DynamoDb) CreateSubscribersTable(ctx context.Context, maxWaitDuration time.Duration) (err error)
  • func (db *DynamoDb) Delete(ctx context.Context, email string) (err error)
  • func (db *DynamoDb) DeleteTable(ctx context.Context) (err error)
  • func (db *DynamoDb) Get(ctx context.Context, email string) (subscriber *Subscriber, err error)
  • func (db *DynamoDb) ProcessSubscribers(ctx context.Context, status SubscriberStatus, sp SubscriberProcessor) error
  • func (db *DynamoDb) Put(ctx context.Context, sub *Subscriber) (err error)
• type DynamoDbClient
• type Subscriber
  • func NewSubscriber(email string) *Subscriber
  • func (sub *Subscriber) String() string
• type SubscriberFunc
  • func (f SubscriberFunc) Process(sub *Subscriber) bool
• type SubscriberProcessor
• type SubscriberStatus

Constants

const DynamoDbPendingIndexName = string(SubscriberPending)

Sparse Global Secondary Index for records containing a "pending" attribute.

const DynamoDbPendingIndexPartitionKey = string(SubscriberPending)

const DynamoDbPrimaryKey = "email"

const DynamoDbVerifiedIndexName string = string(SubscriberVerified)

Sparse Global Secondary Index for records containing a "verified" attribute.

const DynamoDbVerifiedIndexPartitionKey string = string(SubscriberVerified)

const ErrSubscriberNotFound = types.SentinelError("is not a subscriber")

ErrSubscriberNotFound indicates that an email address isn't subscribed.

Database.Get returns this error when the underlying database request succeeded, but there was no such Subscriber.

const TimestampFormat = time.RFC1123Z

Variables

var DynamoDbCreateTableInput = &dynamodb.CreateTableInput{
	AttributeDefinitions: []dbtypes.AttributeDefinition{
		{
			AttributeName: aws.String(DynamoDbPrimaryKey),
			AttributeType: dbtypes.ScalarAttributeTypeS,
		},
		{
			AttributeName: aws.String(DynamoDbPendingIndexPartitionKey),
			AttributeType: dbtypes.ScalarAttributeTypeN,
		},
		{
			AttributeName: aws.String(DynamoDbVerifiedIndexPartitionKey),
			AttributeType: dbtypes.ScalarAttributeTypeN,
		},
	},
	KeySchema: []dbtypes.KeySchemaElement{
		{
			AttributeName: aws.String(DynamoDbPrimaryKey),
			KeyType:       dbtypes.KeyTypeHash,
		},
	},
	BillingMode: dbtypes.BillingModePayPerRequest,
	GlobalSecondaryIndexes: []dbtypes.GlobalSecondaryIndex{
		{
			IndexName: aws.String(DynamoDbPendingIndexName),
			KeySchema: []dbtypes.KeySchemaElement{
				{
					AttributeName: aws.String(DynamoDbPendingIndexPartitionKey),
					KeyType:       dbtypes.KeyTypeHash,
				},
			},
			Projection: DynamoDbIndexProjection,
		},
		{
			IndexName: aws.String(DynamoDbVerifiedIndexName),
			KeySchema: []dbtypes.KeySchemaElement{
				{
					AttributeName: aws.String(
						DynamoDbVerifiedIndexPartitionKey,
					),
					KeyType: dbtypes.KeyTypeHash,
				},
			},
			Projection: DynamoDbIndexProjection,
		},
	},
}

var DynamoDbIndexProjection *dbtypes.Projection = &dbtypes.Projection{
	ProjectionType: dbtypes.ProjectionTypeAll,
}

Types

type Database

type Database interface {
	Get(ctx context.Context, email string) (*Subscriber, error)
	Put(ctx context.Context, subscriber *Subscriber) error
	Delete(ctx context.Context, email string) error
	ProcessSubscribers(
		context.Context, SubscriberStatus, SubscriberProcessor,
	) error
}

type DynamoDb

type DynamoDb struct {
	Client    DynamoDbClient
	TableName string
}

https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/WorkingWithItems.html

func NewDynamoDb

func NewDynamoDb(cfg aws.Config, tableName string) *DynamoDb

func (*DynamoDb) CreateSubscribersTable

func (db *DynamoDb) CreateSubscribersTable(
	ctx context.Context, maxWaitDuration time.Duration,
) (err error)

func (*DynamoDb) Delete

func (db *DynamoDb) Delete(ctx context.Context, email string) (err error)

func (*DynamoDb) DeleteTable

func (db *DynamoDb) DeleteTable(ctx context.Context) (err error)

func (*DynamoDb) Get

func (db *DynamoDb) Get(
	ctx context.Context, email string,
) (subscriber *Subscriber, err error)

func (*DynamoDb) ProcessSubscribers

func (db *DynamoDb) ProcessSubscribers(
	ctx context.Context, status SubscriberStatus, sp SubscriberProcessor,
) error

func (*DynamoDb) Put

func (db *DynamoDb) Put(ctx context.Context, sub *Subscriber) (err error)

type DynamoDbClient

type DynamoDbClient interface {
	CreateTable(
		context.Context, *dynamodb.CreateTableInput, ...func(*dynamodb.Options),
	) (*dynamodb.CreateTableOutput, error)

	DescribeTable(
		context.Context,
		*dynamodb.DescribeTableInput,
		...func(*dynamodb.Options),
	) (*dynamodb.DescribeTableOutput, error)

	UpdateTimeToLive(
		context.Context,
		*dynamodb.UpdateTimeToLiveInput,
		...func(*dynamodb.Options),
	) (*dynamodb.UpdateTimeToLiveOutput, error)

	DeleteTable(
		context.Context, *dynamodb.DeleteTableInput, ...func(*dynamodb.Options),
	) (*dynamodb.DeleteTableOutput, error)

	GetItem(
		context.Context, *dynamodb.GetItemInput, ...func(*dynamodb.Options),
	) (*dynamodb.GetItemOutput, error)

	PutItem(
		context.Context, *dynamodb.PutItemInput, ...func(*dynamodb.Options),
	) (*dynamodb.PutItemOutput, error)

	DeleteItem(
		context.Context, *dynamodb.DeleteItemInput, ...func(*dynamodb.Options),
	) (*dynamodb.DeleteItemOutput, error)

	Scan(
		context.Context, *dynamodb.ScanInput, ...func(*dynamodb.Options),
	) (*dynamodb.ScanOutput, error)
}

type Subscriber

type Subscriber struct {
	Email     string
	Uid       uuid.UUID
	Status    SubscriberStatus
	Timestamp time.Time
}

func NewSubscriber

func NewSubscriber(email string) *Subscriber

func (*Subscriber) String

func (sub *Subscriber) String() string

type SubscriberFunc

type SubscriberFunc func(sub *Subscriber) bool

SubscriberFunc is an adapter to allow processing of Subscriber objects using plain functions.

Inspired by: https://pkg.go.dev/net/http#HandlerFunc

func (SubscriberFunc) Process

func (f SubscriberFunc) Process(sub *Subscriber) bool

SubscriberFunc calls and returns f(sub).

type SubscriberProcessor

type SubscriberProcessor interface {
	Process(*Subscriber) bool
}

A SubscriberProcessor performs an operation on a Subscriber.

Process should return true if processing should continue with the next Subscriber, or false if processing should halt.

type SubscriberStatus

type SubscriberStatus string

const (
	SubscriberPending  SubscriberStatus = "pending"
	SubscriberVerified SubscriberStatus = "verified"
)