persistence

Documentation

Index

• Constants
• Variables
• type ConnectionPoolOptions
• type DBKey
  • func NewDBKey(senderTimestamp uint64, receiverTimestamp uint64, pubsubTopic string, ...) *DBKey
  • func (k *DBKey) Bytes() []byte
• type DBOption
  • func DefaultOptions() []DBOption
  • func WithDB(db *sql.DB) DBOption
  • func WithDriver(driverName string, datasourceName string, ...) DBOption
  • func WithMigrations(migrationFn func(db *sql.DB) error) DBOption
  • func WithRetentionPolicy(maxMessages int, maxDuration time.Duration) DBOption
• type DBStore
  • func NewDBStore(log *zap.Logger, options ...DBOption) (*DBStore, error)
  • func (d *DBStore) Count() (int, error)
  • func (d *DBStore) GetAll() ([]StoredMessage, error)
  • func (d *DBStore) GetStoredMessage(row *sql.Rows) (StoredMessage, error)
  • func (d *DBStore) MostRecentTimestamp() (int64, error)
  • func (d *DBStore) Put(env *protocol.Envelope) error
  • func (d *DBStore) Query(query *pb.HistoryQuery) (*pb.Index, []StoredMessage, error)
  • func (d *DBStore) Start(ctx context.Context, timesource timesource.Timesource) error
  • func (d *DBStore) Stop()
• type Hash
• type MessageProvider
• type StoredMessage

Constants

const (
	TimestampLength   = 8
	HashLength        = 32
	DigestLength      = HashLength
	PubsubTopicLength = HashLength
	DBKeyLength       = TimestampLength + PubsubTopicLength + DigestLength
)

const WALMode = "wal"

WALMode for sqlite.

Variables

var (
	// ErrInvalidByteSize is returned when DBKey can't be created
	// from a byte slice because it has invalid length.
	ErrInvalidByteSize = errors.New("byte slice has invalid length")
)

var ErrInvalidCursor = errors.New("invalid cursor")

Types

type ConnectionPoolOptions

type ConnectionPoolOptions struct {
	MaxOpenConnections    int
	MaxIdleConnections    int
	ConnectionMaxLifetime time.Duration
	ConnectionMaxIdleTime time.Duration
}

type DBKey

type DBKey struct {
	// contains filtered or unexported fields
}

DBKey key to be stored in a db.

func NewDBKey

func NewDBKey(senderTimestamp uint64, receiverTimestamp uint64, pubsubTopic string, digest []byte) *DBKey

NewDBKey creates a new DBKey with the given values.

func (*DBKey) Bytes

func (k *DBKey) Bytes() []byte

Bytes returns a bytes representation of the DBKey.

type DBOption

type DBOption func(*DBStore) error

DBOption is an optional setting that can be used to configure the DBStore

func DefaultOptions

func DefaultOptions() []DBOption

func WithDB

func WithDB(db *sql.DB) DBOption

WithDB is a DBOption that lets you use any custom *sql.DB with a DBStore.

func WithDriver

func WithDriver(driverName string, datasourceName string, connectionPoolOptions ...ConnectionPoolOptions) DBOption

WithDriver is a DBOption that will open a *sql.DB connection

func WithMigrations

func WithMigrations(migrationFn func(db *sql.DB) error) DBOption

WithMigrations is a DBOption used to determine if migrations should be executed, and what driver to use

func WithRetentionPolicy

func WithRetentionPolicy(maxMessages int, maxDuration time.Duration) DBOption

WithRetentionPolicy is a DBOption that specifies the max number of messages to be stored and duration before they're removed from the message store

type DBStore

type DBStore struct {
	MessageProvider
	// contains filtered or unexported fields
}

DBStore is a MessageProvider that has a *sql.DB connection

func NewDBStore

func NewDBStore(log *zap.Logger, options ...DBOption) (*DBStore, error)

Creates a new DB store using the db specified via options. It will create a messages table if it does not exist and clean up records according to the retention policy used

func (*DBStore) Count

func (d *DBStore) Count() (int, error)

Count returns the number of rows in the message table

func (*DBStore) GetAll

func (d *DBStore) GetAll() ([]StoredMessage, error)

GetAll returns all the stored WakuMessages

func (*DBStore) GetStoredMessage

func (d *DBStore) GetStoredMessage(row *sql.Rows) (StoredMessage, error)

GetStoredMessage is a helper function used to convert a `*sql.Rows` into a `StoredMessage`

func (*DBStore) MostRecentTimestamp

func (d *DBStore) MostRecentTimestamp() (int64, error)

MostRecentTimestamp returns an unix timestamp with the most recent senderTimestamp in the message table

func (*DBStore) Put

func (d *DBStore) Put(env *protocol.Envelope) error

Put inserts a WakuMessage into the DB

func (*DBStore) Query

func (d *DBStore) Query(query *pb.HistoryQuery) (*pb.Index, []StoredMessage, error)

Query retrieves messages from the DB

func (*DBStore) Start

func (d *DBStore) Start(ctx context.Context, timesource timesource.Timesource) error

func (*DBStore) Stop

func (d *DBStore) Stop()

Stop closes a DB connection

type Hash

type Hash [HashLength]byte

type MessageProvider

type MessageProvider interface {
	GetAll() ([]StoredMessage, error)
	Put(env *protocol.Envelope) error
	Query(query *pb.HistoryQuery) ([]StoredMessage, error)
	MostRecentTimestamp() (int64, error)
	Start(ctx context.Context, timesource timesource.Timesource) error
	Stop()
}

type StoredMessage

type StoredMessage struct {
	ID           []byte
	PubsubTopic  string
	ReceiverTime int64
	Message      *wpb.WakuMessage
}