Merge branch 'main' into feature/grafana

CONTRIBUTING.md

@@ -5,6 +5,39 @@
 - Go >= 1.17
 - React >= 17.0.0
 
+
+## Fork 代码
+1. 访问 https://github.com/shimohq/mogo
+2. 点击 "Fork" 按钮 (位于页面的右上方)
+
+## Clone 代码
+一般我们推荐将origin设置为官方的仓库，而设置一个自己的upstream。
+
+如果已经在github上开启了SSH，那么我们推荐使用SSH，否则使用HTTPS。两者之间的区别在于，使用HTTPS每次推代码到远程库的时候，都需要输入身份验证信息。
+而我们强烈建议，官方库永远使用HTTPS，这样可以避免一些误操作。
+
+```bash
+git clone https://github.com/shimohq/mogo.git
+cd mogo
+git remote add upstream '[email protected]:<your github username>/mogo.git' 
+```
+upstream可以替换为任何你喜欢的名字。比如说你的用户名，你的昵称，或者直接使用me。后面的命令也要执行相应的替换。
+
+## 同步代码
+除非刚刚把代码拉到本地，否则我们需要先同步一下远程仓库的代码。
+git fetch
+
+在不指定远程库的时候，这个指令只会同步origin的代码。如果我们需要同步自己fork出来的，可以加上远程库名字：
+git fetch upstream
+
+## 创建 feature 分支
+我们在创建新的 feature 分支的时候，要先考虑清楚，从哪个分支切出来。
+我们假设，现在我们希望添加的特性将会被合并到master分支，或者说我们的新特性要在master的基础上进行，执行：
+```bash
+git checkout -b feature/my-feature origin/master
+```
+这样我们就切出来一个分支了。该分支的代码和origin/master上的完全一致。
+
 ### 设置 Api 开发环境
 
 Mogo api 使用 Go 进行开发，如果你本地尚未安装 Go 环境，可以参考[这里](https://go.dev/learn/)安装。

api/internal/apiv1/base/database.go

@@ -0,0 +1,84 @@
+package base
+
+import (
+	"github.com/gotomicro/ego-component/egorm"
+	"github.com/spf13/cast"
+
+	"github.com/shimohq/mogo/api/internal/invoker"
+	"github.com/shimohq/mogo/api/internal/service"
+	"github.com/shimohq/mogo/api/pkg/component/core"
+	"github.com/shimohq/mogo/api/pkg/model/db"
+	"github.com/shimohq/mogo/api/pkg/model/view"
+)
+
+func DatabaseCreate(c *core.Context) {
+	iid := cast.ToInt(c.Param("iid"))
+	if iid == 0 {
+		c.JSONE(core.CodeErr, "invalid parameter", nil)
+		return
+	}
+	var req view.ReqDatabaseCreate
+	if err := c.Bind(&req); err != nil {
+		c.JSONE(1, "invalid parameter: "+err.Error(), nil)
+		return
+	}
+	obj := db.Database{
+		Iid:  iid,
+		Name: req.Name,
+		Uid:  c.Uid(),
+	}
+	op, err := service.InstanceManager.Load(iid)
+	if err != nil {
+		c.JSONE(core.CodeErr, err.Error(), nil)
+		return
+	}
+	err = op.DatabaseCreate(req.Name)
+	if err != nil {
+		c.JSONE(core.CodeErr, "create failed: "+err.Error(), nil)
+		return
+	}
+	if err = db.DatabaseCreate(invoker.Db, &obj); err != nil {
+		c.JSONE(1, "create failed: "+err.Error(), nil)
+		return
+	}
+	c.JSONOK()
+}
+
+func DatabaseList(c *core.Context) {
+	iid := cast.ToInt(c.Param("iid"))
+	conds := egorm.Conds{}
+	if iid != 0 {
+		conds["iid"] = iid
+	}
+	dl, err := db.DatabaseList(invoker.Db, conds)
+	if err != nil {
+		c.JSONE(core.CodeErr, err.Error(), nil)
+		return
+	}
+	res := make([]view.RespDatabaseItem, 0)
+	for _, row := range dl {
+		res = append(res, view.RespDatabaseItem{
+			Id:             row.ID,
+			Iid:            row.Iid,
+			Name:           row.Name,
+			Uid:            row.Uid,
+			DatasourceType: row.Instance.Datasource,
+		})
+	}
+	c.JSONE(core.CodeOK, "succ", res)
+	return
+}
+
+func DatabaseDelete(c *core.Context) {
+	id := cast.ToInt(c.Param("id"))
+	if id == 0 {
+		c.JSONE(1, "invalid parameter", nil)
+		return
+	}
+	err := db.DatabaseDelete(invoker.Db, id)
+	if err != nil {
+		c.JSONE(1, "failed to delete, corresponding record does not exist in database: "+err.Error(), nil)
+		return
+	}
+	c.JSONOK()
+}

api/internal/apiv1/base/table.go

@@ -0,0 +1,222 @@
+package base
+
+import (
+	"github.com/gotomicro/ego-component/egorm"
+	"github.com/spf13/cast"
+
+	"github.com/shimohq/mogo/api/internal/invoker"
+	"github.com/shimohq/mogo/api/internal/service"
+	"github.com/shimohq/mogo/api/pkg/component/core"
+	"github.com/shimohq/mogo/api/pkg/model/db"
+	"github.com/shimohq/mogo/api/pkg/model/view"
+)
+
+func TableId(c *core.Context) {
+	var param view.ReqTableId
+	err := c.Bind(&param)
+	if err != nil {
+		c.JSONE(core.CodeErr, "invalid parameter: "+err.Error(), nil)
+		return
+	}
+	condsIns := egorm.Conds{}
+	condsIns["name"] = param.Instance
+	condsIns["datasource"] = param.Datasource
+	instance, err := db.InstanceInfoX(invoker.Db, condsIns)
+	if err != nil {
+		c.JSONE(core.CodeErr, "invalid parameter: "+err.Error(), nil)
+		return
+	}
+	condsDb := egorm.Conds{}
+	condsDb["iid"] = instance.ID
+	condsDb["name"] = param.Database
+	databaseInfo, err := db.DatabaseInfoX(invoker.Db, condsDb)
+	if err != nil {
+		c.JSONE(core.CodeErr, "invalid parameter: "+err.Error(), nil)
+		return
+	}
+	condsTb := egorm.Conds{}
+	condsTb["did"] = databaseInfo.ID
+	condsTb["name"] = param.Table
+	tableInfo, err := db.TableInfoX(invoker.Db, condsTb)
+	if err != nil {
+		c.JSONE(core.CodeErr, "invalid parameter: "+err.Error(), nil)
+		return
+	}
+	c.JSONOK(tableInfo.ID)
+}
+
+func TableCreate(c *core.Context) {
+	did := cast.ToInt(c.Param("did"))
+	if did == 0 {
+		c.JSONE(core.CodeErr, "params error", nil)
+		return
+	}
+	var param view.ReqTableCreate
+	err := c.Bind(&param)
+	if err != nil {
+		c.JSONE(core.CodeErr, "invalid parameter: "+err.Error(), nil)
+		return
+	}
+	databaseInfo, err := db.DatabaseInfo(invoker.Db, did)
+	if err != nil {
+		c.JSONE(core.CodeErr, "invalid parameter: "+err.Error(), nil)
+		return
+	}
+	op, err := service.InstanceManager.Load(databaseInfo.Iid)
+	if err != nil {
+		c.JSONE(core.CodeErr, err.Error(), nil)
+		return
+	}
+	s, d, v, err := op.TableCreate(did, databaseInfo.Name, param)
+	if err != nil {
+		c.JSONE(core.CodeErr, "create failed: "+err.Error(), nil)
+		return
+	}
+	err = db.TableCreate(invoker.Db, &db.Table{
+		Did:       did,
+		Name:      param.TableName,
+		Typ:       param.Typ,
+		Days:      param.Days,
+		Brokers:   param.Brokers,
+		Topic:     param.Topics,
+		SqlData:   d,
+		SqlStream: s,
+		SqlView:   v,
+		Uid:       c.Uid(),
+	})
+	if err != nil {
+		c.JSONE(core.CodeErr, "create failed: "+err.Error(), nil)
+		return
+	}
+	c.JSONOK()
+}
+
+func TableInfo(c *core.Context) {
+	tid := cast.ToInt(c.Param("id"))
+	if tid == 0 {
+		c.JSONE(core.CodeErr, "params error", nil)
+		return
+	}
+	tableInfo, err := db.TableInfo(invoker.Db, tid)
+	if err != nil {
+		c.JSONE(core.CodeErr, "read list failed: "+err.Error(), nil)
+		return
+	}
+	instance, err := db.InstanceInfo(invoker.Db, tableInfo.Database.Iid)
+	if err != nil {
+		c.JSONE(core.CodeErr, "read list failed: "+err.Error(), nil)
+		return
+	}
+	res := view.RespTableDetail{
+		Did:     tableInfo.Did,
+		Name:    tableInfo.Name,
+		Typ:     tableInfo.Typ,
+		Days:    tableInfo.Days,
+		Brokers: tableInfo.Brokers,
+		Topic:   tableInfo.Topic,
+		Uid:     tableInfo.Uid,
+		Database: view.RespDatabaseItem{
+			Id:             tableInfo.Database.ID,
+			Iid:            tableInfo.Database.Iid,
+			Name:           tableInfo.Database.Name,
+			Uid:            tableInfo.Database.Uid,
+			DatasourceType: instance.Datasource,
+		},
+	}
+	keys := make([]string, 0)
+	data := make(map[string]string, 0)
+	keys = append(keys, "data_sql", "stream_sql", "view_sql")
+	data["data_sql"] = tableInfo.SqlData
+	data["stream_sql"] = tableInfo.SqlStream
+	data["view_sql"] = tableInfo.SqlView
+	conds := egorm.Conds{}
+	conds["tid"] = tableInfo.ID
+	viewList, err := db.ViewList(invoker.Db, conds)
+	if err != nil {
+		c.JSONE(core.CodeErr, "view sql read failed: "+err.Error(), nil)
+		return
+	}
+	for _, v := range viewList {
+		keys = append(keys, v.Name+"_view_sql")
+		data[v.Name+"_view_sql"] = v.SqlView
+	}
+	res.SQLContent.Keys = keys
+	res.SQLContent.Data = data
+	c.JSONOK(res)
+	return
+}
+
+func TableList(c *core.Context) {
+	did := int64(cast.ToInt(c.Param("did")))
+	if did == 0 {
+		c.JSONE(core.CodeErr, "params error", nil)
+		return
+	}
+	conds := egorm.Conds{}
+	conds["did"] = did
+	tableList, err := db.TableList(invoker.Db, conds)
+	if err != nil {
+		c.JSONE(core.CodeErr, "read list failed: "+err.Error(), nil)
+		return
+	}
+	res := make([]view.RespTableSimple, 0)
+	for _, row := range tableList {
+		res = append(res, view.RespTableSimple{
+			Id:        row.ID,
+			TableName: row.Name,
+		})
+	}
+	c.JSONOK(res)
+	return
+}
+
+func TableDelete(c *core.Context) {
+	id := cast.ToInt(c.Param("id"))
+	tableInfo, err := db.TableInfo(invoker.Db, id)
+	if err != nil {
+		c.JSONE(core.CodeErr, "delete failed: "+err.Error(), nil)
+		return
+	}
+	if tableInfo.ID == 0 {
+		c.JSONE(core.CodeErr, "Unable to delete tables not created by Mogo.", nil)
+		return
+	}
+
+	table := tableInfo.Name
+	iid := tableInfo.Database.Iid
+	database := tableInfo.Database.Name
+
+	op, err := service.InstanceManager.Load(iid)
+	if err != nil {
+		c.JSONE(core.CodeErr, err.Error(), nil)
+		return
+	}
+	err = op.TableDrop(database, table, tableInfo.ID)
+	if err != nil {
+		c.JSONE(core.CodeErr, "delete failed: "+err.Error(), nil)
+		return
+	}
+	tx := invoker.Db.Begin()
+	err = db.TableDelete(tx, tableInfo.ID)
+	if err != nil {
+		tx.Rollback()
+		c.JSONE(core.CodeErr, "delete failed: "+err.Error(), nil)
+		return
+	}
+	err = db.ViewDeleteByTableID(tx, tableInfo.ID)
+	if err != nil {
+		tx.Rollback()
+		c.JSONE(core.CodeErr, "delete failed: "+err.Error(), nil)
+		return
+	}
+	err = db.IndexDeleteBatch(tx, tableInfo.ID)
+	if err != nil {
+		tx.Rollback()
+		return
+	}
+	if err = tx.Commit().Error; err != nil {
+		c.JSONE(core.CodeErr, "delete failed: "+err.Error(), nil)
+		return
+	}
+	c.JSONOK("delete succeeded. Note that Kafka may be backlogged.")
+}