优化

accumulate/accumulate.go

@@ -118,10 +118,15 @@ func (a *Accumulation) Evaluate(args any) (result *Result, err error) {
 
 		val := f.Value
 		if f._value != nil {
-			val, err = f._value.EvalString(context.Background(), args)
+			val, err = f._value.EvalFloat64(context.Background(), args)
 			if err != nil {
 				return
 			}
+
+			//路过0值
+			if val.(float64) == 0.0 {
+				continue
+			}
 		}
 
 		ret.Document[key] = val

go.mod

@@ -13,6 +13,7 @@ require (
 	github.com/golang-jwt/jwt/v4 v4.5.0
 	github.com/kardianos/service v1.2.2
 	github.com/panjf2000/ants/v2 v2.10.0
+	github.com/robfig/cron/v3 v3.0.1
 	github.com/santhosh-tekuri/jsonschema/v6 v6.0.1
 	github.com/sirupsen/logrus v1.9.3
 	github.com/spf13/cast v1.6.0

go.sum

@@ -103,6 +103,8 @@ github.com/pelletier/go-toml/v2 v2.2.2/go.mod h1:1t835xjRzz80PqgE6HHgN2JOsmgYu/h
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
+github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/rogpeppe/go-internal v1.11.0 h1:cWPaGQEPrBb5/AsnsZesgZZ9yb1OQ+GOISoDNXVBh4M=
 github.com/rogpeppe/go-internal v1.11.0/go.mod h1:ddIwULY96R17DhadqLgMfk9H9tvdUzkipdSkR5nkCZA=
 github.com/sagikazarmark/locafero v0.6.0 h1:ON7AQg37yzcRPU69mt7gwhFEBwxI6P9T4Qu3N51bwOk=

table/boot.go

@@ -1,16 +1,23 @@
 package table
 
-import "github.com/god-jason/bucket/boot"
+import (
+	"github.com/god-jason/bucket/boot"
+	"github.com/god-jason/bucket/log"
+	"github.com/robfig/cron/v3"
+)
 
 func init() {
 	boot.Register("table", &boot.Task{
 		Startup:  Startup,
-		Shutdown: nil,
+		Shutdown: Shutdown,
 		Depends:  []string{"database", "pool"},
 	})
 }
 
+var _cron *cron.Cron
+
 func Startup() error {
+
 	//加载表定义，编译schema，创建表
 
 	err := LoadAll()
@@ -23,5 +30,27 @@ func Startup() error {
 		return err
 	}
 
+	//定时快照
+	_cron = cron.New()
+	tables.Range(func(name string, item *Table) bool {
+		if item.SnapshotOptions != nil && item.SnapshotOptions.Crontab != "" {
+			_, err := _cron.AddFunc("0 0 * * *", func() {
+				err := item.Snapshot(item.SnapshotOptions.Table)
+				if err != nil {
+					log.Error(err)
+				}
+			})
+			if err != nil {
+				log.Error(err)
+			}
+		}
+		return true
+	})
+	_cron.Start()
+
 	return nil
 }
+
+func Shutdown() error {
+	return _cron.Stop().Err()
+}

table/options.go

@@ -0,0 +1,16 @@
+package table
+
+type SnapshotOptions struct {
+	Crontab string `json:"crontab,omitempty"`
+	Table   string `json:"table,omitempty"`
+}
+
+type DeleteOptions struct {
+	Backup      bool   `json:"backup,omitempty"`
+	BackupTable string `json:"backup_table,omitempty"`
+}
+
+type HistoryOptions struct {
+	Backup      bool   `json:"backup,omitempty"`
+	BackupTable string `json:"backup_table,omitempty"`
+}

table/table.go

@@ -10,6 +10,7 @@ import (
 	"github.com/god-jason/bucket/pkg/javascript"
 	"github.com/santhosh-tekuri/jsonschema/v6"
 	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
 	"go.mongodb.org/mongo-driver/mongo/options"
 	"time"
 )
@@ -33,6 +34,13 @@ type Table struct {
 
 	TimeSeries *options.TimeSeriesOptions `json:"-"` //时间序列参数
 	Hook       *Hook                      `json:"-"`
+
+	//快照
+	SnapshotOptions *SnapshotOptions `json:"snapshot,omitempty"`
+	//备份
+	DeleteOptions *DeleteOptions `json:"delete,omitempty"`
+	//历史
+	HistoryOptions *HistoryOptions `json:"history,omitempty"`
 }
 
 func (t *Table) init() (err error) {
@@ -146,9 +154,12 @@ func (t *Table) Insert(doc any) (id string, err error) {
 			log.Error(err)
 			continue
 		}
-		_, err = db.UpdateMany(ret.Target, ret.Filter, bson.M{"$inc": ret.Document}, true)
-		if err != nil {
-			log.Error(err)
+
+		if len(ret.Document) > 0 {
+			_, err = db.UpdateMany(ret.Target, ret.Filter, bson.M{"$inc": ret.Document}, true)
+			if err != nil {
+				log.Error(err)
+			}
 		}
 	}
 
@@ -246,8 +257,18 @@ func (t *Table) Delete(id string) error {
 		return NotFound
 	}
 
-	//把删除保存到修改历史表
-	_, _ = db.InsertOne(t.Name+".deleted", result)
+	//备份
+	if t.DeleteOptions != nil && t.DeleteOptions.Backup {
+		tab := t.DeleteOptions.BackupTable
+		if tab == "" {
+			tab = t.Name + ".history"
+		}
+		//把删除保存到修改历史表
+		result["__id"] = oid
+		result["deleted"] = time.Now()
+		delete(result, "_id")
+		_, _ = db.InsertOne(tab, result)
+	}
 
 	//转换_id
 	db.StringifyDocumentObjectId(result)
@@ -276,9 +297,11 @@ func (t *Table) Delete(id string) error {
 			log.Error(err)
 			continue
 		}
-		_, err = db.UpdateMany(ret.Target, ret.Filter, bson.M{"$dec": ret.Document}, false)
-		if err != nil {
-			log.Error(err)
+		if len(ret.Document) > 0 {
+			_, err = db.UpdateMany(ret.Target, ret.Filter, bson.M{"$dec": ret.Document}, false)
+			if err != nil {
+				log.Error(err)
+			}
 		}
 	}
 
@@ -334,6 +357,19 @@ func (t *Table) Update(id string, update any) error {
 	//把差异保存到修改历史表
 	_, _ = db.InsertOne(t.Name+".change", bson.M{"object_id": oid, "base": base, "change": update})
 
+	//备份
+	if t.HistoryOptions != nil && t.HistoryOptions.Backup {
+		tab := t.HistoryOptions.BackupTable
+		if tab == "" {
+			tab = t.Name + ".history"
+		}
+		//把删除保存到修改历史表
+		base["__id"] = oid
+		base["updated"] = time.Now()
+		delete(base, "_id")
+		_, _ = db.InsertOne(tab, base)
+	}
+
 	//转换—_id
 	db.StringifyDocumentObjectId(update)
 	db.StringifyDocumentObjectId(base)
@@ -363,9 +399,11 @@ func (t *Table) Update(id string, update any) error {
 			log.Error(err)
 			continue
 		}
-		_, err = db.UpdateMany(ret.Target, ret.Filter, bson.M{"$dec": ret.Document}, true)
-		if err != nil {
-			log.Error(err)
+		if len(ret.Document) > 0 {
+			_, err = db.UpdateMany(ret.Target, ret.Filter, bson.M{"$dec": ret.Document}, true)
+			if err != nil {
+				log.Error(err)
+			}
 		}
 	}
 	//补充字段，base已经被污染
@@ -380,9 +418,11 @@ func (t *Table) Update(id string, update any) error {
 			log.Error(err)
 			continue
 		}
-		_, err = db.UpdateMany(ret.Target, ret.Filter, bson.M{"$inc": ret.Document}, true)
-		if err != nil {
-			log.Error(err)
+		if len(ret.Document) > 0 {
+			_, err = db.UpdateMany(ret.Target, ret.Filter, bson.M{"$inc": ret.Document}, true)
+			if err != nil {
+				log.Error(err)
+			}
 		}
 	}
 
@@ -441,3 +481,21 @@ func (t *Table) DistinctId(filter any) (values []string, err error) {
 	}
 	return values, nil
 }
+
+func (t *Table) Snapshot(into string) (err error) {
+	var docs []db.Document
+
+	//默认表名
+	if into == "" {
+		into = t.Name + ".snapshot"
+	}
+
+	now := time.Now()
+	pipeline := mongo.Pipeline{
+		bson.D{{"$set", bson.M{"object_id": "$_id", "date": now}}},
+		bson.D{{"$unset", "_id"}},
+		bson.D{{"$merge", bson.M{"into": into}}},
+	}
+
+	return t.Aggregate(pipeline, &docs)
+}