fix(storage,document): sequential upserts

cmd/cli/command/app/package.go

@@ -125,23 +125,34 @@ func copyDir(writer *zip.Writer, baseDir string, zipBasePath string) error {
 }
 
 func copyFile(writer *zip.Writer, srcPath string, zipPath string) error {
-	r, err := os.Open(srcPath)
+	srcFile, err := os.Open(srcPath)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	srcStat, err := os.Stat(srcPath)
 	if err != nil {
 		return errors.WithStack(err)
 	}
 
 	defer func() {
-		if err := r.Close(); err != nil {
+		if err := srcFile.Close(); err != nil {
 			panic(errors.WithStack(err))
 		}
 	}()
 
-	f, err := writer.Create(zipPath)
+	fileHeader := &zip.FileHeader{
+		Name:     zipPath,
+		Modified: srcStat.ModTime().UTC(),
+		Method:   zip.Deflate,
+	}
+
+	file, err := writer.CreateHeader(fileHeader)
 	if err != nil {
 		return errors.WithStack(err)
 	}
 
-	if _, err = io.Copy(f, r); err != nil {
+	if _, err = io.Copy(file, srcFile); err != nil {
 		return errors.WithStack(err)
 	}
 

pkg/http/options.go

@@ -26,7 +26,7 @@ func defaultHandlerOptions() *HandlerOptions {
 		Bus:                   memory.NewBus(),
 		SockJS:                sockjsOptions,
 		ServerModuleFactories: make([]app.ServerModuleFactory, 0),
-		UploadMaxFileSize:     1024 * 10, // 10Mb
+		UploadMaxFileSize:     10 << (10 * 2), // 10Mb
 	}
 }
 

pkg/storage/document_store.go

@@ -31,12 +31,17 @@ func (d Document) ID() (DocumentID, bool) {
 		return "", false
 	}
 
-	id, ok := rawID.(string)
+	strID, ok := rawID.(string)
 	if ok {
-		return "", false
+		return DocumentID(strID), true
 	}
 
-	return DocumentID(id), true
+	docID, ok := rawID.(DocumentID)
+	if ok {
+		return docID, true
+	}
+
+	return "", false
 }
 
 func (d Document) CreatedAt() (time.Time, bool) {
@@ -54,7 +59,7 @@ func (d Document) timeAttr(attr string) (time.Time, bool) {
 	}
 
 	t, ok := rawTime.(time.Time)
-	if ok {
+	if !ok {
 		return time.Time{}, false
 	}
 

pkg/storage/filter/sql/operator.go

@@ -0,0 +1,15 @@
+package sql
+
+const (
+	OpIn                = "IN"
+	OpLesserThan        = "<"
+	OpLesserThanEqual   = "<="
+	OpEqual             = "="
+	OpNotEqual          = "!="
+	OpSuperiorThan      = ">"
+	OpSuperiorThanEqual = ">="
+	OpAnd               = "AND"
+	OpOr                = "OR"
+	OpLike              = "LIKE"
+	OpNot               = "NOT"
+)

pkg/storage/filter/sql/option.go

@@ -71,6 +71,12 @@ func WithDefaultTransform() OptionFunc {
 	}
 }
 
+func WithTransform(transform TransformFunc) OptionFunc {
+	return func(opt *Option) {
+		opt.Transform = transform
+	}
+}
+
 func WithNoOpValueTransform() OptionFunc {
 	return WithValueTransform(func(value interface{}) interface{} {
 		return value

pkg/storage/filter/sql/sql.go

@@ -60,7 +60,7 @@ func transformAndOperator(op filter.Operator, option *Option) (string, []interfa
 		return "", nil, errors.Wrapf(filter.ErrUnexpectedOperator, "expected '%s', got '%s'", filter.TokenAnd, op.Token())
 	}
 
-	return aggregatorToSQL("AND", option, andOp.Children()...)
+	return aggregatorToSQL(OpAnd, option, andOp.Children()...)
 }
 
 func transformOrOperator(op filter.Operator, option *Option) (string, []interface{}, error) {
@@ -69,7 +69,7 @@ func transformOrOperator(op filter.Operator, option *Option) (string, []interfac
 		return "", nil, errors.Wrapf(filter.ErrUnexpectedOperator, "expected '%s', got '%s'", filter.TokenOr, op.Token())
 	}
 
-	return aggregatorToSQL("OR", option, orOp.Children()...)
+	return aggregatorToSQL(OpOr, option, orOp.Children()...)
 }
 
 func transformEqOperator(op filter.Operator, option *Option) (string, []interface{}, error) {
@@ -78,7 +78,7 @@ func transformEqOperator(op filter.Operator, option *Option) (string, []interfac
 		return "", nil, errors.Wrapf(filter.ErrUnexpectedOperator, "expected '%s', got '%s'", filter.TokenEq, op.Token())
 	}
 
-	return fieldsToSQL("=", false, eqOp.Fields(), option)
+	return fieldsToSQL(OpEqual, false, eqOp.Fields(), option)
 }
 
 func transformNeqOperator(op filter.Operator, option *Option) (string, []interface{}, error) {
@@ -87,7 +87,7 @@ func transformNeqOperator(op filter.Operator, option *Option) (string, []interfa
 		return "", nil, errors.Wrapf(filter.ErrUnexpectedOperator, "expected '%s', got '%s'", filter.TokenNeq, op.Token())
 	}
 
-	return fieldsToSQL("!=", false, eqOp.Fields(), option)
+	return fieldsToSQL(OpNotEqual, false, eqOp.Fields(), option)
 }
 
 func transformGtOperator(op filter.Operator, option *Option) (string, []interface{}, error) {
@@ -96,7 +96,7 @@ func transformGtOperator(op filter.Operator, option *Option) (string, []interfac
 		return "", nil, errors.Wrapf(filter.ErrUnexpectedOperator, "expected '%s', got '%s'", filter.TokenGt, op.Token())
 	}
 
-	return fieldsToSQL(">", false, gtOp.Fields(), option)
+	return fieldsToSQL(OpSuperiorThan, false, gtOp.Fields(), option)
 }
 
 func transformGteOperator(op filter.Operator, option *Option) (string, []interface{}, error) {
@@ -105,7 +105,7 @@ func transformGteOperator(op filter.Operator, option *Option) (string, []interfa
 		return "", nil, errors.Wrapf(filter.ErrUnexpectedOperator, "expected '%s', got '%s'", filter.TokenGte, op.Token())
 	}
 
-	return fieldsToSQL(">=", false, gteOp.Fields(), option)
+	return fieldsToSQL(OpSuperiorThanEqual, false, gteOp.Fields(), option)
 }
 
 func transformLtOperator(op filter.Operator, option *Option) (string, []interface{}, error) {
@@ -114,7 +114,7 @@ func transformLtOperator(op filter.Operator, option *Option) (string, []interfac
 		return "", nil, errors.Wrapf(filter.ErrUnexpectedOperator, "expected '%s', got '%s'", filter.TokenLt, op.Token())
 	}
 
-	return fieldsToSQL("<", false, ltOp.Fields(), option)
+	return fieldsToSQL(OpLesserThan, false, ltOp.Fields(), option)
 }
 
 func transformLteOperator(op filter.Operator, option *Option) (string, []interface{}, error) {
@@ -123,7 +123,7 @@ func transformLteOperator(op filter.Operator, option *Option) (string, []interfa
 		return "", nil, errors.Wrapf(filter.ErrUnexpectedOperator, "expected '%s', got '%s'", filter.TokenLte, op.Token())
 	}
 
-	return fieldsToSQL("<=", false, lteOp.Fields(), option)
+	return fieldsToSQL(OpLesserThanEqual, false, lteOp.Fields(), option)
 }
 
 func transformInOperator(op filter.Operator, option *Option) (string, []interface{}, error) {
@@ -132,7 +132,7 @@ func transformInOperator(op filter.Operator, option *Option) (string, []interfac
 		return "", nil, errors.Wrapf(filter.ErrUnexpectedOperator, "expected '%s', got '%s'", filter.TokenIn, op.Token())
 	}
 
-	return fieldsToSQL("IN", true, inOp.Fields(), option)
+	return fieldsToSQL(OpIn, true, inOp.Fields(), option)
 }
 
 func transformLikeOperator(op filter.Operator, option *Option) (string, []interface{}, error) {
@@ -141,7 +141,7 @@ func transformLikeOperator(op filter.Operator, option *Option) (string, []interf
 		return "", nil, errors.Wrapf(filter.ErrUnexpectedOperator, "expected '%s', got '%s'", filter.TokenLike, op.Token())
 	}
 
-	return fieldsToSQL("LIKE", false, likeOp.Fields(), option)
+	return fieldsToSQL(OpLike, false, likeOp.Fields(), option)
 }
 
 func transformNotOperator(op filter.Operator, option *Option) (string, []interface{}, error) {
@@ -150,10 +150,10 @@ func transformNotOperator(op filter.Operator, option *Option) (string, []interfa
 		return "", nil, errors.Wrapf(filter.ErrUnexpectedOperator, "expected '%s', got '%s'", filter.TokenNot, op.Token())
 	}
 
-	sql, args, err := aggregatorToSQL("AND", option, notOp.Children()...)
+	sql, args, err := aggregatorToSQL(OpAnd, option, notOp.Children()...)
 	if err != nil {
 		return "", nil, errors.WithStack(err)
 	}
 
-	return "NOT " + sql, args, nil
+	return OpNot + " " + sql, args, nil
 }

pkg/storage/filter/sql/transform.go

@@ -32,7 +32,7 @@ func DefaultTransform(operator string, invert bool, key string, value interface{
 			return "", nil, errors.WithStack(err)
 		}
 
-		if _, err := sb.WriteString(key); err != nil {
+		if _, err := sb.WriteString(option.KeyTransform(key)); err != nil {
 			return "", nil, errors.WithStack(err)
 		}
 	} else {

pkg/storage/sqlite/document_store.go

@@ -102,6 +102,7 @@ func (s *DocumentStore) Query(ctx context.Context, collection string, filter *fi
 			criteria, args, err = filterSQL.ToSQL(
 				filter.Root(),
 				filterSQL.WithPreparedParameter("$", 2),
+				filterSQL.WithTransform(transformOperator),
 				filterSQL.WithKeyTransform(func(key string) string {
 					return fmt.Sprintf("json_extract(data, '$.%s')", key)
 				}),
@@ -187,10 +188,6 @@ func (s *DocumentStore) Upsert(ctx context.Context, collection string, document
 			id = storage.NewDocumentID()
 		}
 
-		delete(document, storage.DocumentAttrID)
-		delete(document, storage.DocumentAttrCreatedAt)
-		delete(document, storage.DocumentAttrUpdatedAt)
-
 		args := []any{id, collection, JSONMap(document), now, now}
 
 		row := tx.QueryRowContext(ctx, query, args...)

pkg/storage/sqlite/filter.go

@@ -0,0 +1,24 @@
+package sqlite
+
+import (
+	"fmt"
+
+	"forge.cadoles.com/arcad/edge/pkg/storage/filter/sql"
+)
+
+func transformOperator(operator string, invert bool, key string, value any, option *sql.Option) (string, any, error) {
+	switch operator {
+	case sql.OpIn:
+		return transformInOperator(key, value, option)
+	default:
+		return sql.DefaultTransform(operator, invert, key, value, option)
+	}
+}
+
+func transformInOperator(key string, value any, option *sql.Option) (string, any, error) {
+	return fmt.Sprintf(
+		"EXISTS (SELECT 1 FROM json_each(json_extract(data, \"$.%v\")) WHERE value = %v)",
+		key,
+		option.PreparedParameter(),
+	), option.ValueTransform(value), nil
+}

pkg/storage/testsuite/document_store.go

@@ -7,8 +7,8 @@ import (
 )
 
 func TestDocumentStore(t *testing.T, store storage.DocumentStore) {
-	t.Run("Query", func(t *testing.T) {
+	t.Run("Ops", func(t *testing.T) {
 		// t.Parallel()
-		testDocumentStoreQuery(t, store)
+		testDocumentStoreOps(t, store)
 	})
 }

pkg/storage/testsuite/document_store_ops.go

@@ -0,0 +1,212 @@
+package testsuite
+
+import (
+	"context"
+	"testing"
+
+	"forge.cadoles.com/arcad/edge/pkg/storage"
+	"forge.cadoles.com/arcad/edge/pkg/storage/filter"
+	"github.com/davecgh/go-spew/spew"
+	"github.com/pkg/errors"
+)
+
+type documentStoreOpsTestCase struct {
+	Name string
+	Run  func(ctx context.Context, store storage.DocumentStore) error
+}
+
+var documentStoreOpsTestCases = []documentStoreOpsTestCase{
+	{
+		Name: "Basic query",
+		Run: func(ctx context.Context, store storage.DocumentStore) error {
+			collection := "simple_select"
+
+			docs := []storage.Document{
+				{
+					"attr1": "Foo",
+				},
+				{
+					"attr1": "Bar",
+				},
+			}
+
+			for _, d := range docs {
+				if _, err := store.Upsert(ctx, collection, d); err != nil {
+					return errors.WithStack(err)
+				}
+			}
+
+			filter := filter.New(
+				filter.NewEqOperator(map[string]interface{}{
+					"attr1": "Foo",
+				}),
+			)
+
+			results, err := store.Query(ctx, collection, filter, nil)
+			if err != nil {
+				return errors.WithStack(err)
+			}
+
+			if e, g := 1, len(results); e != g {
+				return errors.Errorf("len(results): expected '%v', got '%v'", e, g)
+			}
+
+			if e, g := "Foo", results[0]["attr1"]; e != g {
+				return errors.Errorf("results[0][\"Attr1\"]: expected '%v', got '%v'", e, g)
+			}
+
+			return nil
+		},
+	},
+	{
+		Name: "Query with 'IN' operator",
+		Run: func(ctx context.Context, store storage.DocumentStore) error {
+			docs := []storage.Document{
+				{
+					"counter": 1,
+					"tags":    []string{"foo", "bar"},
+				},
+				{
+					"counter": 1,
+					"tags":    []string{"nope"},
+				},
+			}
+
+			collection := "in_operator"
+
+			for _, doc := range docs {
+				if _, err := store.Upsert(ctx, collection, doc); err != nil {
+					return errors.WithStack(err)
+				}
+			}
+
+			filter := filter.New(
+				filter.NewAndOperator(
+					filter.NewEqOperator(map[string]any{
+						"counter": 1,
+					}),
+					filter.NewInOperator(map[string]any{
+						"tags": "foo",
+					}),
+				),
+			)
+
+			results, err := store.Query(ctx, collection, filter, nil)
+			if err != nil {
+				return errors.WithStack(err)
+			}
+
+			if e, g := 1, len(results); e != g {
+				return errors.Errorf("len(results): expected '%v', got '%v'", e, g)
+			}
+
+			return nil
+		},
+	},
+	{
+		Name: "Double upsert",
+		Run: func(ctx context.Context, store storage.DocumentStore) error {
+			collection := "double_upsert"
+
+			oriDoc := storage.Document{
+				"attr1": "Foo",
+			}
+
+			// Upsert document for the first time
+			upsertedDoc, err := store.Upsert(ctx, collection, oriDoc)
+			if err != nil {
+				return errors.WithStack(err)
+			}
+
+			id, exists := upsertedDoc.ID()
+			if !exists {
+				return errors.New("id, exists := upsertedDoc.ID(): 'exists' should be true")
+			}
+
+			if id == storage.DocumentID("") {
+				return errors.New("id, exists := upsertedDoc.ID(): 'id' should not be an empty string")
+			}
+
+			createdAt, exists := upsertedDoc.CreatedAt()
+			if !exists {
+				return errors.New("createdAt, exists := upsertedDoc.CreatedAt(): 'exists' should be true")
+			}
+
+			if createdAt.IsZero() {
+				return errors.New("createdAt, exists := upsertedDoc.CreatedAt(): 'createdAt' should not be zero time")
+			}
+
+			updatedAt, exists := upsertedDoc.UpdatedAt()
+			if !exists {
+				return errors.New("updatedAt, exists := upsertedDoc.UpdatedAt(): 'exists' should be true")
+			}
+
+			if updatedAt.IsZero() {
+				return errors.New("updatedAt, exists := upsertedDoc.UpdatedAt(): 'updatedAt' should not be zero time")
+			}
+
+			if e, g := oriDoc["attr1"], upsertedDoc["attr1"]; e != g {
+				return errors.Errorf("upsertedDoc[\"attr1\"]: expected '%v', got '%v'", e, g)
+			}
+
+			// Check that document does not have unexpected properties
+			if e, g := 4, len(upsertedDoc); e != g {
+				return errors.Errorf("len(upsertedDoc): expected '%v', got '%v'", e, g)
+			}
+
+			// Upsert document for the second time
+			upsertedDoc2, err := store.Upsert(ctx, collection, upsertedDoc)
+			if err != nil {
+				return errors.WithStack(err)
+			}
+
+			spew.Dump(upsertedDoc, upsertedDoc2)
+
+			prevID, _ := upsertedDoc.ID()
+			newID, _ := upsertedDoc2.ID()
+
+			if e, g := prevID, newID; e != g {
+				return errors.Errorf("newID: expected '%v', got '%v'", e, g)
+			}
+
+			createdAt1, _ := upsertedDoc.CreatedAt()
+			createdAt2, _ := upsertedDoc2.CreatedAt()
+
+			if e, g := createdAt1, createdAt2; e != g {
+				return errors.Errorf("upsertedDoc2.CreatedAt(): expected '%v', got '%v'", e, g)
+			}
+
+			updatedAt1, _ := upsertedDoc.UpdatedAt()
+			updatedAt2, _ := upsertedDoc2.UpdatedAt()
+
+			if e, g := updatedAt1, updatedAt2; e == g {
+				return errors.New("upsertedDoc2.UpdatedAt() should have been different than upsertedDoc.UpdatedAt()")
+			}
+
+			// Verify that there is no additional created document in the collection
+
+			results, err := store.Query(ctx, collection, nil, nil)
+			if err != nil {
+				return errors.WithStack(err)
+			}
+
+			if e, g := 1, len(results); e != g {
+				return errors.Errorf("len(results): expected '%v', got '%v'", e, g)
+			}
+
+			return nil
+		},
+	},
+}
+
+func testDocumentStoreOps(t *testing.T, store storage.DocumentStore) {
+	for _, tc := range documentStoreOpsTestCases {
+		func(tc documentStoreOpsTestCase) {
+			t.Run(tc.Name, func(t *testing.T) {
+				if err := tc.Run(context.Background(), store); err != nil {
+					t.Errorf("%+v", errors.WithStack(err))
+				}
+			})
+		}(tc)
+	}
+}

pkg/storage/testsuite/document_store_query.go

@@ -1,85 +0,0 @@
-package testsuite
-
-import (
-	"context"
-	"testing"
-
-	"forge.cadoles.com/arcad/edge/pkg/storage"
-	"forge.cadoles.com/arcad/edge/pkg/storage/filter"
-	"github.com/pkg/errors"
-)
-
-type documentStoreQueryTestCase struct {
-	Name              string
-	Before            func(ctx context.Context, store storage.DocumentStore) error
-	Collection        string
-	Filter            *filter.Filter
-	QueryOptionsFuncs []storage.QueryOptionFunc
-	After             func(t *testing.T, results []storage.Document, err error)
-}
-
-var documentStoreQueryTestCases = []documentStoreQueryTestCase{
-	{
-		Name: "Simple select",
-		Before: func(ctx context.Context, store storage.DocumentStore) error {
-			doc1 := storage.Document{
-				"attr1": "Foo",
-			}
-
-			if _, err := store.Upsert(ctx, "simple_select", doc1); err != nil {
-				return errors.WithStack(err)
-			}
-
-			doc2 := storage.Document{
-				"attr1": "Bar",
-			}
-
-			if _, err := store.Upsert(ctx, "simple_select", doc2); err != nil {
-				return errors.WithStack(err)
-			}
-
-			return nil
-		},
-		Collection: "simple_select",
-		Filter: filter.New(
-			filter.NewEqOperator(map[string]interface{}{
-				"attr1": "Foo",
-			}),
-		),
-		After: func(t *testing.T, results []storage.Document, err error) {
-			if err != nil {
-				t.Fatalf("%+v", errors.WithStack(err))
-			}
-
-			if e, g := 1, len(results); e != g {
-				t.Errorf("len(results): expected '%v', got '%v'", e, g)
-			}
-
-			if e, g := "Foo", results[0]["attr1"]; e != g {
-				t.Errorf("results[0][\"Attr1\"]: expected '%v', got '%v'", e, g)
-			}
-		},
-	},
-}
-
-func testDocumentStoreQuery(t *testing.T, store storage.DocumentStore) {
-	for _, tc := range documentStoreQueryTestCases {
-		func(tc documentStoreQueryTestCase) {
-			t.Run(tc.Name, func(t *testing.T) {
-				// t.Parallel()
-
-				ctx := context.Background()
-
-				if tc.Before != nil {
-					if err := tc.Before(ctx, store); err != nil {
-						t.Fatalf("%+v", errors.WithStack(err))
-					}
-				}
-
-				documents, err := store.Query(ctx, tc.Collection, tc.Filter, tc.QueryOptionsFuncs...)
-
-				tc.After(t, documents, err)
-			})
-		}(tc)
-	}
-}