WIP: document store

Author: alash3al

commands_doc.go.wip

@@ -0,0 +1,216 @@
+package main
+
+import (
+	"encoding/csv"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/alash3al/redix/kvstore"
+	"github.com/jeremywohl/flatten"
+	"github.com/rs/xid"
+	"github.com/tidwall/gjson"
+	"github.com/tidwall/sjson"
+)
+
+func dsetCommand(c Context) {
+	if len(c.args) < 2 {
+		c.WriteError("DSET command requires at least 2 arguments: DSET <collection> [<json> <json> ...]")
+		return
+	}
+
+	collection := c.args[0]
+
+	docs := map[string]string{}
+	indexes := map[string]string{}
+
+	for _, j := range c.args[1:] {
+		if !gjson.Valid(j) {
+			continue
+		}
+
+		randID := xid.New().String()
+		parseddoc := gjson.Parse(j)
+		docid := ""
+
+		if !parseddoc.Get("id").Exists() || parseddoc.Get("id").String() == "" {
+			j, _ = sjson.Set(j, "id", randID)
+
+			docid = randID
+		} else {
+			docid = parseddoc.Get("id").String()
+		}
+
+		if !parseddoc.Get("created_at").Exists() {
+			j, _ = sjson.Set(j, "created_at", time.Now().UnixNano())
+		}
+
+		j, _ = sjson.Set(j, "updated_at", time.Now().UnixNano())
+
+		flatdoc, err := flatten.FlattenString(j, "", flatten.DotStyle)
+		if err != nil {
+			continue
+		}
+
+		docs[collection+"/{DOCUMENT}/{RAW}/"+docid] = j
+
+		gjson.Parse(flatdoc).ForEach(func(k, v gjson.Result) bool {
+			// <collection>/{DOCUMENT}/{INDEX}/<docfield>/<docfieldvalue>/<sequence> = <docid>
+			indexes[collection+"/{DOCUMENT}/{INDEX}/"+k.String()+"/"+v.Raw+"/"+xid.New().String()] = docid
+			return true
+		})
+	}
+
+	c.db.MSet(docs)
+	c.db.MSet(indexes)
+
+	c.WriteArray(len(docs) * 2)
+	for k, v := range docs {
+		k = strings.SplitN(k, "/{DOCUMENT}/{RAW}/", 2)[1]
+
+		c.WriteBulkString(k)
+		c.WriteBulkString(v)
+	}
+}
+
+func dgetCommand(c Context) {
+	if len(c.args) < 2 {
+		c.WriteError("DGET requires at least 2 arguments: DGET <collection> <docid>")
+		return
+	}
+
+	collection, docid := c.args[0], c.args[1]
+
+	doc, err := c.db.Get(collection + "/{DOCUMENT}/{RAW}/" + docid)
+	if err != nil {
+		c.WriteError(err.Error())
+		return
+	}
+
+	c.WriteBulkString(doc)
+}
+
+func dgetallCommand(c Context) {
+	if len(c.args) < 1 {
+		c.WriteError("DGETALL command requires at least one argument: DGETALL <collection> [<offset> <limit>]")
+		return
+	}
+
+	prefix := c.args[0] + "/{DOCUMENT}/{RAW}/"
+	offset := prefix
+	limit := 5
+
+	if len(c.args) > 1 && c.args[1] != "" {
+		offset = prefix + c.args[1]
+	}
+
+	if len(c.args) > 2 {
+		limit, _ = strconv.Atoi(c.args[2])
+	}
+
+	data := []string{}
+	length := 0
+	err := c.db.Scan(kvstore.ScannerOptions{
+		FetchValues:   true,
+		IncludeOffset: offset == prefix,
+		Prefix:        prefix,
+		Offset:        offset,
+		Handler: func(k, v string) bool {
+			length++
+			p := strings.SplitN(k, "/{DOCUMENT}/{RAW}/", 2)
+			if len(p) < 2 {
+				return true
+			}
+			data = append(data, p[1], v)
+			if length >= limit {
+				return false
+			}
+			return true
+		},
+	})
+
+	if err != nil {
+		c.WriteError(err.Error())
+		return
+	}
+
+	c.WriteArray(len(data))
+
+	for _, v := range data {
+		c.WriteBulkString(v)
+	}
+}
+
+func dfilterCommand(c Context) {
+	if len(c.args) < 2 {
+		c.WriteError("DFILTER command requires at least one argument: DFILTER <collection> <filter> [<offset> <limit>]")
+		return
+	}
+
+	prefix := c.args[0] + "/{DOCUMENT}/{INDEX}/"
+	filters, err := (func() ([]string, error) {
+		r := csv.NewReader(strings.NewReader(c.args[1]))
+		r.Comma = ' '
+		return r.Read()
+	})()
+
+	if err != nil {
+		c.WriteError(err.Error())
+		return
+	}
+
+	offset := prefix
+	limit := 5
+
+	if len(c.args) > 1 && c.args[1] != "" {
+		offset = prefix + c.args[1]
+	}
+
+	if len(c.args) > 2 {
+		limit, _ = strconv.Atoi(c.args[2])
+	}
+
+	data := []string{}
+	length := 0
+
+	// all := map[string][]string{}
+	wg := sync.WaitGroup{}
+
+	for _, filter := range filters {
+		parts := strings.SplitN(filter, ":", 2)
+		if len(parts) < 2 {
+			continue
+		}
+		wg.Add(1)
+		go (func() {
+			defer wg.Done()
+			c.db.Scan(kvstore.ScannerOptions{
+				FetchValues:   true,
+				IncludeOffset: offset == prefix,
+				Prefix:        prefix,
+				Offset:        offset,
+				Handler: func(k, v string) bool {
+					length++
+					p := strings.SplitN(k, "/{DOCUMENT}/{RAW}/", 2)
+					if len(p) < 2 {
+						return true
+					}
+					data = append(data, p[1], v)
+					if length >= limit {
+						return false
+					}
+					return true
+				},
+			})
+		})()
+	}
+
+	wg.Wait()
+
+	c.WriteArray(len(data))
+
+	for _, v := range data {
+		c.WriteBulkString(v)
+	}
+}

commands_strings.go

@@ -26,6 +26,9 @@ func setCommand(c Context) {
 	}
 
 	ttlVal, _ := strconv.Atoi(ttl)
+	if ttlVal < 0 {
+		ttlVal = 0
+	}
 
 	if err := c.db.Set(k, v, ttlVal); err != nil {
 		c.WriteError(err.Error())

helpers.go helpers_db.gohelpers.go renamed to helpers_db.go

@@ -8,11 +8,10 @@ import (
 	"path/filepath"
 	"strings"
 
-	"github.com/alash3al/redix/kvstore/bolt"
-
-	"github.com/alash3al/redix/kvstore/badger"
-
 	"github.com/alash3al/redix/kvstore"
+	"github.com/alash3al/redix/kvstore/badger"
+	"github.com/alash3al/redix/kvstore/bolt"
+	"github.com/alash3al/redix/kvstore/level"
 )
 
 // selectDB - load/fetches the requested db
@@ -41,5 +40,7 @@ func openDB(engine, dbpath string) (kvstore.DB, error) {
 		return badger.OpenBadger(dbpath)
 	case "bolt", "boltdb":
 		return bolt.OpenBolt(dbpath)
+	case "level", "leveldb":
+		return level.OpenLevelDB(dbpath)
 	}
 }

init.go

@@ -36,7 +36,7 @@ func init() {
 		return
 	}
 
-	color.Cyan(redixBrand)
+	color.Blue(redixBrand)
 
 	databases = new(sync.Map)
 	changelog = pubsub.NewBroker()
@@ -53,7 +53,9 @@ func init() {
 		if !f.IsDir() {
 			continue
 		}
+
 		name := filepath.Base(f.Name())
+
 		_, err := selectDB(name)
 		if err != nil {
 			log.Println(color.RedString(err.Error()))