Impl delete find filters for boltdb

2020-07-23 16:41:50 -06:00
parent 085150a7b5
commit bed265a228
13 changed files with 670 additions and 92 deletions
--- a/storage/driver/boltdb.go
+++ b/storage/driver/boltdb.go
@@ -0,0 +1,172 @@
+package driver
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"local/dndex/config"
+	"os"
+	"regexp"
+
+	"github.com/boltdb/bolt"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/bson/primitive"
+)
+
+type BoltDB struct {
+	db *bolt.DB
+}
+
+func NewBoltDB() *BoltDB {
+	config := config.New()
+	db, err := bolt.Open(config.DBURI, os.ModePerm, nil)
+	if err != nil {
+		panic(err)
+	}
+	return &BoltDB{
+		db: db,
+	}
+}
+
+func (bdb *BoltDB) count(ctx context.Context, namespace string, filter interface{}) (int, error) {
+	ch, err := bdb.Find(ctx, namespace, filter)
+	n := 0
+	for _ = range ch {
+		n += 1
+	}
+	return n, err
+}
+
+func (bdb *BoltDB) Find(_ context.Context, namespace string, filter interface{}) (chan bson.Raw, error) {
+	b, err := bson.Marshal(filter)
+	if err != nil {
+		return nil, err
+	}
+	m := bson.M{}
+	if err := bson.Unmarshal(b, &m); err != nil {
+		return nil, err
+	}
+	results := make([]bson.Raw, 0)
+	err = bdb.db.View(func(tx *bolt.Tx) error {
+		bucket := tx.Bucket([]byte(namespace))
+		if bucket == nil {
+			return nil
+		}
+		cursor := bucket.Cursor()
+		for k, v := cursor.First(); k != nil && v != nil; k, v = cursor.Next() {
+			n := bson.M{}
+			if err := bson.Unmarshal(v, &n); err != nil {
+				return err
+			}
+			if matches(n, m) {
+				results = append(results, bson.Raw(v))
+			}
+		}
+		return nil
+	})
+	ch := make(chan bson.Raw)
+	go func() {
+		defer close(ch)
+		for i := range results {
+			ch <- results[i]
+		}
+	}()
+	return ch, err
+}
+
+func (bdb *BoltDB) Update(context.Context, string, interface{}, interface{}) error {
+	return errors.New("not impl")
+}
+
+func (bdb *BoltDB) Insert(context.Context, string, interface{}) error {
+	return errors.New("not impl")
+}
+
+func (bdb *BoltDB) Delete(_ context.Context, namespace string, filter interface{}) error {
+	b, err := bson.Marshal(filter)
+	if err != nil {
+		return err
+	}
+	m := bson.M{}
+	if err := bson.Unmarshal(b, &m); err != nil {
+		return err
+	}
+	return bdb.db.Update(func(tx *bolt.Tx) error {
+		bucket := tx.Bucket([]byte(namespace))
+		if bucket == nil {
+			return nil
+		}
+		cursor := bucket.Cursor()
+		for k, v := cursor.First(); k != nil && v != nil; k, v = cursor.Next() {
+			n := bson.M{}
+			if err := bson.Unmarshal(v, &n); err != nil {
+				return err
+			}
+			if matches(n, m) {
+				if err := bucket.Delete(k); err != nil {
+					return err
+				}
+			}
+		}
+		return nil
+	})
+}
+
+func matches(doc, filter bson.M) bool {
+	for k, v := range filter {
+		if _, ok := doc[k]; !ok {
+			continue
+		}
+		switch v.(type) {
+		case map[string]interface{}, primitive.M:
+			m, ok := v.(map[string]interface{})
+			if !ok {
+				m = map[string]interface{}(v.(primitive.M))
+			}
+			for k2, v2 := range m {
+				switch k2 {
+				case "$regex":
+					pattern, ok := v2.(string)
+					if !ok {
+						return false
+					}
+					re, err := regexp.Compile(pattern)
+					if err != nil {
+						return false
+					}
+					if !re.MatchString(fmt.Sprint(doc[k])) {
+						return false
+					}
+				case "$in":
+					options, ok := v2.([]interface{})
+					if !ok {
+						options = []interface{}(v2.(primitive.A))
+						ok = true
+					}
+					matches := false
+					for _, option := range options {
+						if fmt.Sprint(doc[k]) == fmt.Sprint(option) {
+							matches = true
+						}
+					}
+					if !matches {
+						return false
+					}
+				default:
+					dock, ok := doc[k].(map[string]interface{})
+					if !ok {
+						return false
+					}
+					if !matches(dock, bson.M(m)) {
+						return false
+					}
+				}
+			}
+		default:
+			if fmt.Sprint(v) != fmt.Sprint(doc[k]) {
+				return false
+			}
+		}
+	}
+	return true
+}