👌 feat(message:deduplication) implementing the feature #33

gbus/abstractions.go

@@ -22,6 +22,14 @@ const (
 	REPLY Semantics = "reply"
 )
 
+type DeduplicationPolicy int
+
+const (
+	DeduplicationPolicyNone DeduplicationPolicy = iota
+	DeduplicationPolicyReject
+	DeduplicationPolicyAck
+)
+
 //Bus interface provides the majority of functionality to Send, Reply and Publish messages to the Bus
 type Bus interface {
 	HandlerRegister
@@ -213,6 +221,8 @@ type Builder interface {
 
 	//WithLogger set custom logger instance
 	WithLogger(logger logrus.FieldLogger) Builder
+
+	WithDeduplicationPolicy(method DeduplicationPolicy, age time.Duration) Builder
 }
 
 //Invocation context for a specific processed message

gbus/builder/builder.go

@@ -5,31 +5,35 @@ import (
 	"sync"
 	"time"
 
+	"emperror.dev/errors"
 	"github.com/sirupsen/logrus"
 
 	"github.com/wework/grabbit/gbus"
+	"github.com/wework/grabbit/gbus/deduplicator/implementation"
 	"github.com/wework/grabbit/gbus/saga"
 	"github.com/wework/grabbit/gbus/serialization"
 	"github.com/wework/grabbit/gbus/tx/mysql"
 )
 
 type defaultBuilder struct {
-	PrefetchCount    uint
-	connStr          string
-	purgeOnStartup   bool
-	sagaStoreConnStr string
-	txnl             bool
-	txConnStr        string
-	txnlProvider     string
-	workerNum        uint
-	serializer       gbus.Serializer
-	dlx              string
-	defaultPolicies  []gbus.MessagePolicy
-	confirm          bool
-	dbPingTimeout    time.Duration
-	usingPingTimeout bool
-	logger           logrus.FieldLogger
-	busCfg           gbus.BusConfiguration
+	PrefetchCount             uint
+	connStr                   string
+	purgeOnStartup            bool
+	sagaStoreConnStr          string
+	txnl                      bool
+	txConnStr                 string
+	txnlProvider              string
+	workerNum                 uint
+	serializer                gbus.Serializer
+	dlx                       string
+	defaultPolicies           []gbus.MessagePolicy
+	confirm                   bool
+	dbPingTimeout             time.Duration
+	usingPingTimeout          bool
+	logger                    logrus.FieldLogger
+	busCfg                    gbus.BusConfiguration
+	deduplicationPolicy       gbus.DeduplicationPolicy
+	deduplicationRetentionAge time.Duration
 }
 
 func (builder *defaultBuilder) Build(svcName string) gbus.Bus {
@@ -53,6 +57,7 @@ func (builder *defaultBuilder) Build(svcName string) gbus.Bus {
 		DefaultPolicies:      builder.defaultPolicies,
 		DbPingTimeout:        3,
 		Confirm:              builder.confirm,
+		DeduplicationPolicy:  builder.deduplicationPolicy,
 	}
 
 	var finalLogger logrus.FieldLogger
@@ -107,6 +112,7 @@ func (builder *defaultBuilder) Build(svcName string) gbus.Bus {
 	if builder.usingPingTimeout {
 		gb.DbPingTimeout = builder.dbPingTimeout
 	}
+	gb.Deduplicator = implementation.NewDeduplicator(svcName, builder.deduplicationPolicy, gb.TxProvider, builder.deduplicationRetentionAge, gb.Log())
 
 	//TODO move this into the NewSagaStore factory methods
 	if builder.purgeOnStartup {
@@ -115,6 +121,11 @@ func (builder *defaultBuilder) Build(svcName string) gbus.Bus {
 			errMsg := fmt.Errorf("grabbit: saga store faild to purge. error: %v", err)
 			panic(errMsg)
 		}
+		err = gb.Deduplicator.Purge()
+		if err != nil {
+			errMsg := errors.NewWithDetails("duplicator failed to purge", "component", "grabbit", "feature", "deduplicator")
+			panic(errMsg)
+		}
 	}
 	glue := saga.NewGlue(gb, sagaStore, svcName, gb.TxProvider, gb.Log, timeoutManager)
 	glue.SetLogger(gb.Log())
@@ -206,6 +217,12 @@ func (builder *defaultBuilder) WithLogger(logger logrus.FieldLogger) gbus.Builde
 	return builder
 }
 
+func (builder *defaultBuilder) WithDeduplicationPolicy(policy gbus.DeduplicationPolicy, age time.Duration) gbus.Builder {
+	builder.deduplicationPolicy = policy
+	builder.deduplicationRetentionAge = age
+	return builder
+}
+
 //New :)
 func New() Nu {
 	return Nu{}
@@ -218,9 +235,11 @@ type Nu struct {
 //Bus inits a new BusBuilder
 func (Nu) Bus(brokerConnStr string) gbus.Builder {
 	return &defaultBuilder{
-		busCfg:          gbus.BusConfiguration{},
-		PrefetchCount:   1,
-		connStr:         brokerConnStr,
-		serializer:      serialization.NewGobSerializer(),
-		defaultPolicies: make([]gbus.MessagePolicy, 0)}
+		busCfg:              gbus.BusConfiguration{},
+		PrefetchCount:       1,
+		connStr:             brokerConnStr,
+		serializer:          serialization.NewGobSerializer(),
+		defaultPolicies:     make([]gbus.MessagePolicy, 0),
+		deduplicationPolicy: gbus.DeduplicationPolicyNone,
+	}
 }

gbus/bus.go

@@ -9,6 +9,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/wework/grabbit/gbus/deduplicator"
 	"github.com/wework/grabbit/gbus/metrics"
 
 	"github.com/opentracing-contrib/go-amqp/amqptracer"
@@ -56,15 +57,17 @@ type DefaultBus struct {
 	Glue                 SagaGlue
 	TxProvider           TxProvider
 
-	WorkerNum       uint
-	Serializer      Serializer
-	DLX             string
-	DefaultPolicies []MessagePolicy
-	Confirm         bool
-	healthChan      chan error
-	backpressure    bool
-	DbPingTimeout   time.Duration
-	amqpConnected   bool
+	WorkerNum           uint
+	Serializer          Serializer
+	DLX                 string
+	DeduplicationPolicy DeduplicationPolicy
+	Deduplicator        deduplicator.Store
+	DefaultPolicies     []MessagePolicy
+	Confirm             bool
+	healthChan          chan error
+	backpressure        bool
+	DbPingTimeout       time.Duration
+	amqpConnected       bool
 }
 
 var (
@@ -222,6 +225,8 @@ func (b *DefaultBus) Start() error {
 		return startErr
 	}
 
+	b.Deduplicator.Start()
+
 	//declare queue
 	var q amqp.Queue
 	if q, e = b.createServiceQueue(); e != nil {
@@ -294,7 +299,10 @@ func (b *DefaultBus) createBusWorkers(workerNum uint) ([]*worker, error) {
 			registrations:     b.Registrations,
 			serializer:        b.Serializer,
 			b:                 b,
-			amqpErrors:        b.amqpErrors}
+			amqpErrors:        b.amqpErrors,
+			delicatePolicy:    b.DeduplicationPolicy,
+			duplicateStore:    b.Deduplicator,
+		}
 
 		err := w.Start()
 		if err != nil {

gbus/deduplicator/implementation/tx.go

@@ -0,0 +1,138 @@
+package implementation
+
+import (
+	"database/sql"
+	"time"
+
+	"emperror.dev/errors"
+	"github.com/sirupsen/logrus"
+
+	"github.com/wework/grabbit/gbus"
+	"github.com/wework/grabbit/gbus/deduplicator"
+	"github.com/wework/grabbit/gbus/tx"
+)
+
+var _ deduplicator.Store = &deduper{}
+
+type deduper struct {
+	*gbus.Glogged
+	svcName    string
+	policy     gbus.DeduplicationPolicy
+	txProvider gbus.TxProvider
+	age        time.Duration
+	ticker     *time.Ticker
+	done       chan bool
+	tableName  string
+}
+
+func (d *deduper) Purge() (err error) {
+	truncateSQL := "TRUNCATE TABLE " + d.tableName
+	txp, err := d.txProvider.New()
+	if err != nil {
+		return err
+	}
+	defer func() {
+		if err != nil {
+			serr := txp.Rollback()
+			err = errors.Append(err, serr)
+		}
+		err = txp.Commit()
+	}()
+	_, err = txp.Exec(truncateSQL)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (d *deduper) Start() {
+	d.ticker = time.NewTicker(time.Minute)
+	d.done = make(chan bool)
+	deleteQuery := "DELETE FROM " + d.tableName + " WHERE `created_at` < ?"
+	go func() {
+		for {
+			select {
+			case <-d.done:
+				return
+			case <-d.ticker.C:
+				oldest := time.Now().Add(-1 * d.age)
+				tx, err := d.txProvider.New()
+				if err != nil {
+					d.Log().WithError(err).Error("failed to acquire a tx")
+					continue
+				}
+				result, err := tx.Exec(deleteQuery, oldest)
+				if err != nil && err != sql.ErrNoRows {
+					d.Log().WithError(err).Error("failed executing delete query")
+				}
+				n, err := result.RowsAffected()
+				if err != nil {
+					d.Log().WithError(err).Error("failed to get count of affected rows")
+				} else {
+					d.Log().WithField("table_name", d.tableName).WithField("rows_deleted", n).
+						Info("successfully cleanup duplicates table")
+				}
+			}
+		}
+	}()
+}
+
+func (d *deduper) Stop() {
+	d.Log().Info("shutting down deduplicator")
+	d.ticker.Stop()
+	close(d.done)
+}
+
+//
+func (d *deduper) StoreMessageID(tx *sql.Tx, id string) error {
+	insertSQL := "INSERT INTO " + d.tableName + " (id) values (?)"
+	_, err := tx.Exec(insertSQL, id)
+	if err != nil {
+		d.Log().WithError(err).Error("failed to insert the id of the message into the dedup table")
+		return err
+	}
+	return nil
+}
+
+// MessageExists checks if a message id is in the deduplication table and returns an error if it fails
+func (d *deduper) MessageExists(id string) (bool, error) {
+	if d.policy == gbus.DeduplicationPolicyNone {
+		return false, nil
+	}
+	tx, err := d.txProvider.New()
+	if err != nil {
+		return true, err
+	}
+	defer func() {
+		err = tx.Rollback()
+		if err != nil {
+			d.Log().WithError(err).Error("could not commit tx for query MessageExists")
+		}
+	}()
+	selectSQL := "SELECT EXISTS (SELECT id FROM " + d.tableName + " WHERE id = ? limit 1)"
+
+	var exists bool
+	err = tx.QueryRow(selectSQL, id).Scan(&exists)
+	if err != nil && err == sql.ErrNoRows {
+		return false, nil
+	}
+
+	if err != nil {
+		return true, err
+	}
+
+	return exists, nil
+}
+
+func NewDeduplicator(svcName string, policy gbus.DeduplicationPolicy, txProvider gbus.TxProvider, age time.Duration, logger logrus.FieldLogger) deduplicator.Store {
+	d := &deduper{
+		svcName:    svcName,
+		policy:     policy,
+		txProvider: txProvider,
+		age:        age,
+		tableName:  tx.GrabbitTableNameTemplate(svcName, "duplicates"),
+	}
+	l := logger.WithField("grabbit", "deduplicator")
+	d.SetLogger(l)
+	return d
+}

gbus/deduplicator/store.go

@@ -0,0 +1,14 @@
+package deduplicator
+
+import (
+	"database/sql"
+)
+
+// Store abstracts the way deduplicateor manages the
+type Store interface {
+	StoreMessageID(tx *sql.Tx, id string) error
+	MessageExists(id string) (bool, error)
+	Purge() error
+	Start()
+	Stop()
+}