chore: implement configurable deduplication mirror mode (#6224)

# Description

## Overview
Refactored the dedup package to use a single Dedup instance with
pluggable database backends instead of creating multiple instances.

## Key Changes

### 1. Core Architecture
- **Before**: Multiple Dedup instances (BadgerDB, KeyDB, Mirror)
- **After**: Single Dedup instance with pluggable DB backends

### 2. Interface Changes
- Added `DB` interface in `types` package for database operations
- Made BadgerDB and KeyDB implement the DB interface
- Created MirrorDB to handle mirroring logic internally

### 3. Main Dedup Structure
```go
// New simplified structure
type Dedup struct {
    db           DB
    uncommittedMu sync.RWMutex
    uncommitted  map[string]struct{}
}
```

### 4. Database Implementations

#### BadgerDB (`services/dedup/badger/badger.go`)
- `NewBadgerDB` now returns `types.DB` instead of `*Dedup`
- Removed embedded Dedup struct
- Implements DB interface directly

#### KeyDB (`services/dedup/keydb/keydb.go`)
- `NewKeyDB` now returns `types.DB` instead of `*Dedup`
- Removed embedded Dedup struct
- Implements DB interface directly

#### MirrorDB (`services/dedup/mirror.go`)
- New implementation that wraps primary and mirror databases
- Handles asynchronous mirroring operations
- Implements both DB and DedupInterface interfaces

### 5. Configuration Modes
1. `badgerOnlyMode` - BadgerDB only
2. `keyDBOnlyMode` - KeyDB only
3. `mirrorBadgerMode` - BadgerDB primary, KeyDB mirror
4. `mirrorKeyDBMode` - KeyDB primary, BadgerDB mirror

### 6. Test Updates (`services/dedup/dedup_test.go`)
- Updated to use `types.DedupInterface` instead of `dedup.Dedup`
- Modified KeyDB test to use config to select backend
- Preserved all existing test logic and coverage

## Benefits
- Single Dedup instance management
- Cleaner separation of concerns
- Easier testing and maintenance
- Better extensibility for new database backends
- Preserved backward compatibility

## Linear Ticket

pipe-2289

## Security

- [x] The code changed/added as part of this pull request won't create
any security issues with how the software is being used.

Author: mihir20

mocks/services/dedup/mock_dedup.go

@@ -15,6 +15,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/rudderlabs/rudder-server/services/dedup"
+
 	"github.com/google/uuid"
 	"github.com/samber/lo"
 	"github.com/tidwall/gjson"
@@ -46,7 +48,7 @@ import (
 	"github.com/rudderlabs/rudder-server/rruntime"
 	destinationdebugger "github.com/rudderlabs/rudder-server/services/debugger/destination"
 	transformationdebugger "github.com/rudderlabs/rudder-server/services/debugger/transformation"
-	"github.com/rudderlabs/rudder-server/services/dedup"
+	deduptypes "github.com/rudderlabs/rudder-server/services/dedup/types"
 	"github.com/rudderlabs/rudder-server/services/fileuploader"
 	"github.com/rudderlabs/rudder-server/services/rmetrics"
 	"github.com/rudderlabs/rudder-server/services/rsources"
@@ -113,7 +115,7 @@ type Handle struct {
 	pendingEventsRegistry      rmetrics.PendingEventsRegistry
 	logger                     logger.Logger
 	enrichers                  []enricher.PipelineEnricher
-	dedup                      dedup.Dedup
+	dedup                      deduptypes.Dedup
 	reporting                  reportingtypes.Reporting
 	reportingEnabled           bool
 	backgroundWait             func() error

processor/processor.go

@@ -9,29 +9,27 @@ import (
 	"time"
 
 	obskit "github.com/rudderlabs/rudder-observability-kit/go/labels"
+	"github.com/rudderlabs/rudder-server/rruntime"
 
 	"github.com/dgraph-io/badger/v4"
 	"github.com/dgraph-io/badger/v4/options"
-	"github.com/samber/lo"
 
 	"github.com/rudderlabs/rudder-go-kit/bytesize"
 	"github.com/rudderlabs/rudder-go-kit/config"
 	"github.com/rudderlabs/rudder-go-kit/logger"
 	"github.com/rudderlabs/rudder-go-kit/stats"
 
-	"github.com/rudderlabs/rudder-server/rruntime"
 	"github.com/rudderlabs/rudder-server/services/dedup/types"
 	"github.com/rudderlabs/rudder-server/utils/misc"
 )
 
-type BadgerDB struct {
+type badgerDB struct {
 	logger           loggerForBadger
 	badgerDB         *badger.DB
 	window           config.ValueLoader[time.Duration]
 	path             string
 	opts             badger.Options
 	cleanupOnStartup bool
-	once             sync.Once
 
 	wg     sync.WaitGroup
 	bgCtx  context.Context
@@ -55,7 +53,7 @@ func DefaultPath() string {
 	return fmt.Sprintf(`%v%v`, tmpDirPath, badgerPathName)
 }
 
-func NewBadgerDB(conf *config.Config, stat stats.Stats, path string) *Dedup {
+func NewBadgerDB(conf *config.Config, stat stats.Stats, path string) (types.DB, error) {
 	dedupWindow := conf.GetReloadableDurationVar(3600, time.Second, "Dedup.dedupWindow", "Dedup.dedupWindowInS")
 	log := logger.NewLogger().Child("Dedup")
 	badgerOpts := badger.
@@ -81,7 +79,7 @@ func NewBadgerDB(conf *config.Config, stat stats.Stats, path string) *Dedup {
 		WithDetectConflicts(conf.GetBoolVar(false, "BadgerDB.Dedup.detectConflicts", "BadgerDB.detectConflicts"))
 
 	bgCtx, cancel := context.WithCancel(context.Background())
-	db := &BadgerDB{
+	db := &badgerDB{
 		logger:           loggerForBadger{log},
 		path:             path,
 		window:           dedupWindow,
@@ -96,13 +94,14 @@ func NewBadgerDB(conf *config.Config, stat stats.Stats, path string) *Dedup {
 	db.stats.vlogSize = stat.NewTaggedStat("badger_db_size", stats.GaugeType, stats.Tags{"name": "dedup", "type": "vlog"})
 	db.stats.totSize = stat.NewTaggedStat("badger_db_size", stats.GaugeType, stats.Tags{"name": "dedup", "type": "total"})
 
-	return &Dedup{
-		badgerDB:    db,
-		uncommitted: make(map[string]struct{}),
+	err := db.init()
+	if err != nil {
+		return nil, fmt.Errorf("initializing badger db: %w", err)
 	}
+	return db, nil
 }
 
-func (d *BadgerDB) Get(keys []string) (map[string]bool, error) {
+func (d *badgerDB) Get(keys []string) (map[string]bool, error) {
 	defer d.stats.getTimer.RecordDuration()()
 	results := make(map[string]bool, len(keys))
 	err := d.badgerDB.View(func(txn *badger.Txn) error {
@@ -120,7 +119,7 @@ func (d *BadgerDB) Get(keys []string) (map[string]bool, error) {
 	return results, err
 }
 
-func (d *BadgerDB) Set(keys []string) error {
+func (d *badgerDB) Set(keys []string) error {
 	defer d.stats.setTimer.RecordDuration()()
 	wb := d.badgerDB.NewWriteBatch()
 	defer wb.Cancel()
@@ -133,57 +132,55 @@ func (d *BadgerDB) Set(keys []string) error {
 	return wb.Flush()
 }
 
-func (d *BadgerDB) Close() {
+func (d *badgerDB) Close() {
 	d.cancel()
 	d.wg.Wait()
 	if d.badgerDB != nil {
 		_ = d.badgerDB.Close()
 	}
 }
 
-func (d *BadgerDB) init() error {
+func (d *badgerDB) init() error {
 	var err error
-	d.once.Do(func() {
-		if d.cleanupOnStartup {
-			if err = os.RemoveAll(d.path); err != nil {
-				err = fmt.Errorf("removing badger db directory: %w", err)
-				return
-			}
+	if d.cleanupOnStartup {
+		if err = os.RemoveAll(d.path); err != nil {
+			err = fmt.Errorf("removing badger db directory: %w", err)
+			return err
 		}
-		openDB := func() (dbase *badger.DB, err error) {
-			defer func() {
-				if r := recover(); r != nil {
-					err = fmt.Errorf("panic during badgerdb open: %v", r)
-				}
-			}()
-			return badger.Open(d.opts)
+	}
+	openDB := func() (dbase *badger.DB, err error) {
+		defer func() {
+			if r := recover(); r != nil {
+				err = fmt.Errorf("panic during badgerdb open: %v", r)
+			}
+		}()
+		return badger.Open(d.opts)
+	}
+	d.badgerDB, err = openDB()
+	if err != nil {
+		// corrupted or incompatible db, clean up the directory and retry
+		d.logger.Errorn("Error while opening dedup badger db, cleaning up the directory",
+			obskit.Error(err),
+		)
+		if err = os.RemoveAll(d.opts.Dir); err != nil {
+			err = fmt.Errorf("removing badger db directory: %w", err)
+			return err
 		}
 		d.badgerDB, err = openDB()
 		if err != nil {
-			// corrupted or incompatible db, clean up the directory and retry
-			d.logger.Errorn("Error while opening dedup badger db, cleaning up the directory",
-				obskit.Error(err),
-			)
-			if err = os.RemoveAll(d.opts.Dir); err != nil {
-				err = fmt.Errorf("removing badger db directory: %w", err)
-				return
-			}
-			d.badgerDB, err = openDB()
-			if err != nil {
-				err = fmt.Errorf("opening badger db: %w", err)
-				return
-			}
+			err = fmt.Errorf("opening badger db: %w", err)
+			return err
 		}
-		d.wg.Add(1)
-		rruntime.Go(func() {
-			defer d.wg.Done()
-			d.gcLoop()
-		})
+	}
+	d.wg.Add(1)
+	rruntime.Go(func() {
+		defer d.wg.Done()
+		d.gcLoop()
 	})
 	return err
 }
 
-func (d *BadgerDB) gcLoop() {
+func (d *badgerDB) gcLoop() {
 	for {
 		select {
 		case <-d.bgCtx.Done():
@@ -213,87 +210,6 @@ func (d *BadgerDB) gcLoop() {
 	}
 }
 
-type Dedup struct {
-	badgerDB      *BadgerDB
-	uncommittedMu sync.RWMutex
-	uncommitted   map[string]struct{}
-}
-
-func (d *Dedup) Allowed(batchKeys ...types.BatchKey) (map[types.BatchKey]bool, error) {
-	if err := d.badgerDB.init(); err != nil {
-		return nil, fmt.Errorf("initializing badger db: %w", err)
-	}
-	result := make(map[types.BatchKey]bool, len(batchKeys))  // keys encountered for the first time
-	seenInBatch := make(map[string]struct{}, len(batchKeys)) // keys already seen in the batch while iterating
-
-	// figure out which keys need to be checked against the DB
-	batchKeysToCheck := make([]types.BatchKey, 0, len(batchKeys)) // keys to check in the DB
-	d.uncommittedMu.RLock()
-	for _, batchKey := range batchKeys {
-		// if the key is already seen in the batch, skip it
-		if _, seen := seenInBatch[batchKey.Key]; seen {
-			continue
-		}
-		// if the key is already in the uncommitted list , skip it
-		if _, uncommitted := d.uncommitted[batchKey.Key]; uncommitted {
-			seenInBatch[batchKey.Key] = struct{}{}
-			continue
-		}
-		seenInBatch[batchKey.Key] = struct{}{}
-		batchKeysToCheck = append(batchKeysToCheck, batchKey)
-	}
-	d.uncommittedMu.RUnlock()
-
-	if len(batchKeysToCheck) > 0 {
-		seenInDB, err := d.badgerDB.Get(lo.Map(batchKeysToCheck, func(bk types.BatchKey, _ int) string { return bk.Key }))
-		if err != nil {
-			return nil, fmt.Errorf("getting keys from badger db: %w", err)
-		}
-		d.uncommittedMu.Lock()
-		defer d.uncommittedMu.Unlock()
-		for _, batchKey := range batchKeysToCheck {
-			if !seenInDB[batchKey.Key] {
-				if _, race := d.uncommitted[batchKey.Key]; !race { // if another goroutine managed to set this key, we should skip it
-					result[batchKey] = true
-					d.uncommitted[batchKey.Key] = struct{}{} // mark this key as uncommitted
-				}
-			}
-		}
-	}
-	return result, nil
-}
-
-func (d *Dedup) Commit(keys []string) error {
-	if err := d.badgerDB.init(); err != nil {
-		return fmt.Errorf("initializing badger db: %w", err)
-	}
-	kvs := make([]types.BatchKey, len(keys))
-	d.uncommittedMu.RLock()
-	for i, key := range keys {
-		if _, ok := d.uncommitted[key]; !ok {
-			d.uncommittedMu.RUnlock()
-			return fmt.Errorf("key %v has not been previously set", key)
-		}
-		kvs[i] = types.BatchKey{Key: key}
-	}
-	d.uncommittedMu.RUnlock()
-
-	if err := d.badgerDB.Set(keys); err != nil {
-		return fmt.Errorf("setting keys in badger db: %w", err)
-	}
-
-	d.uncommittedMu.Lock()
-	defer d.uncommittedMu.Unlock()
-	for _, kv := range kvs {
-		delete(d.uncommitted, kv.Key)
-	}
-	return nil
-}
-
-func (d *Dedup) Close() {
-	d.badgerDB.Close()
-}
-
 type loggerForBadger struct {
 	logger.Logger
 }