本地存储

 package main
 
 import (
-	"encoding/json"
 	"flag"
-	"fmt"
 	"github.com/golang-jwt/jwt/v4/request"
 	"github.com/zeromicro/go-queue/kq"
 	"github.com/zeromicro/go-zero/core/logx"
@@ -69,26 +67,26 @@ func main() {
 func startConsume(c config.Config) {
 	//svcCtx := svc.NewServiceContext(c)
 	go func() {
-		for {
-			notice := &domain.ObjectNotice{
-				CompanyId:         1598224576532189184,
-				TableId:           521,
-				TableType:         "主表",
-				ObjectType:        "导入模块",
-				Event:             "table.data.edit",
-				TableAffectedList: []int{521},
-				DataChanged:       true,
-				StructChanged:     true,
-				MetaData: domain.ObjectNoticeMetaData{
-					Module: 0,
-					Status: 0,
-				},
-			}
-			mBytes, _ := json.Marshal(notice)
-			err := kq.NewPusher(c.KqConsumerConf.Brokers, c.KqConsumerConf.Topic).Push(string(mBytes))
-			fmt.Println(err)
-			time.Sleep(10 * 10 * time.Second)
-		}
+		//for {
+		//	notice := &domain.ObjectNotice{
+		//		CompanyId:         1598224576532189184,
+		//		TableId:           521,
+		//		TableType:         "主表",
+		//		ObjectType:        "导入模块",
+		//		Event:             "table.data.edit",
+		//		TableAffectedList: []int{521},
+		//		DataChanged:       true,
+		//		StructChanged:     true,
+		//		MetaData: domain.ObjectNoticeMetaData{
+		//			Module: 0,
+		//			Status: 0,
+		//		},
+		//	}
+		//	mBytes, _ := json.Marshal(notice)
+		//	err := kq.NewPusher(c.KqConsumerConf.Brokers, c.KqConsumerConf.Topic).Push(string(mBytes))
+		//	fmt.Println(err)
+		//	time.Sleep(10 * 10 * time.Second)
+		//}
 	}()
 	//kafka消费队列 处理字库推送事件
 	go func() {

@@ -33,7 +33,7 @@ ByteMetadata:
 KqConsumerConf:
   Name: KqConsumer
   Brokers:
-    - 192.168.100.221:9092
-  Group: sumifcc
-  Topic: sumifcc
+    - 47.97.5.102:9092
+  Group: bchart_dev
+  Topic: allied_creation_metadata_table_sync_notice
   Processors: 1

@@ -95,6 +95,11 @@ func RegisterHandlers(server *rest.Server, serverCtx *svc.ServiceContext) {
 				Path:    "/table/data",
 				Handler: table.SearchTableDataHandler(serverCtx),
 			},
+			{
+				Method:  http.MethodGet,
+				Path:    "/table/sync",
+				Handler: table.SyncTableHandler(serverCtx),
+			},
 		},
 		rest.WithPrefix("/v1"),
 	)

+package table
+
+import (
+	"net/http"
+
+	"github.com/zeromicro/go-zero/rest/httpx"
+	"gitlab.fjmaimaimai.com/allied-creation/sumifcc-bchart/cmd/chart-server/api/internal/logic/table"
+	"gitlab.fjmaimaimai.com/allied-creation/sumifcc-bchart/cmd/chart-server/api/internal/svc"
+)
+
+func SyncTableHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		l := table.NewSyncTableLogic(r.Context(), svcCtx)
+		resp, err := l.SyncTable()
+		if err != nil {
+			httpx.ErrorCtx(r.Context(), w, err)
+		} else {
+			httpx.OkJsonCtx(r.Context(), w, resp)
+		}
+	}
+}

@@ -133,8 +133,8 @@ func (l *SearchTableByModuleLogic) getTableByLocal(modules []string) (types.Sear
 		List:  make([]types.SearchTableByModuleItem, 0),
 	}
 	total, list, err := l.svcCtx.ObjectTableRepository.Find(l.ctx, l.svcCtx.DefaultDBConn(), map[string]interface{}{
-		"companyId": tenantId,
-		"tableType": modules,
+		"companyId":   tenantId,
+		"tableTypeIn": modules,
 	})
 	if err != nil {
 		return response, err

+package table
+
+import (
+	"context"
+	"gitlab.fjmaimaimai.com/allied-creation/sumifcc-bchart/cmd/chart-server/api/internal/types"
+	"gitlab.fjmaimaimai.com/allied-creation/sumifcc-bchart/cmd/chart-server/interanl/pkg/db/transaction"
+	"gitlab.fjmaimaimai.com/allied-creation/sumifcc-bchart/cmd/chart-server/interanl/pkg/domain"
+	"gitlab.fjmaimaimai.com/allied-creation/sumifcc-bchart/cmd/chart-server/interanl/pkg/gateway/bytelib"
+	"gitlab.fjmaimaimai.com/allied-creation/sumifcc-bchart/pkg/xerr"
+
+	"github.com/zeromicro/go-zero/core/logx"
+	"gitlab.fjmaimaimai.com/allied-creation/sumifcc-bchart/cmd/chart-server/api/internal/svc"
+)
+
+type SyncTableLogic struct {
+	logx.Logger
+	ctx    context.Context
+	svcCtx *svc.ServiceContext
+	conn   transaction.Conn
+}
+
+func NewSyncTableLogic(ctx context.Context, svcCtx *svc.ServiceContext) *SyncTableLogic {
+	return &SyncTableLogic{
+		Logger: logx.WithContext(ctx),
+		ctx:    ctx,
+		svcCtx: svcCtx,
+		conn:   svcCtx.DefaultDBConn(),
+	}
+}
+
+func (l *SyncTableLogic) SyncTable() (resp interface{}, err error) {
+	companyIds, err := l.svcCtx.ChartRepository.FindCompanyIds(l.ctx, l.conn)
+	if err != nil {
+		return nil, xerr.NewErrMsg("查询公司ID失败：" + err.Error())
+	}
+	tables := make([]*domain.ObjectTable, 0)
+	for _, companyId := range companyIds {
+		accessToken, _ := types.TableAccessToken{CompanyId: companyId}.GenerateToken()
+		//获取导入模块
+		response, err := l.svcCtx.ByteMetadataService.ObjectTableSearch(l.ctx, bytelib.ObjectTableSearchRequest{
+			Token:      accessToken,
+			TableTypes: []string{bytelib.MainTable, bytelib.SubTable, bytelib.SideTable},
+			Module:     bytelib.ModuleChartTemplate,
+		})
+		if err == nil {
+			tables = append(tables, l.getTables(companyId, response)...)
+		}
+		//拆解模块
+		response, err = l.svcCtx.ByteMetadataService.ObjectTableSearch(l.ctx, bytelib.ObjectTableSearchRequest{
+			Token:           accessToken,
+			TableTypes:      []string{bytelib.SchemaTable},
+			Module:          bytelib.ModuleQuerySetCenter,
+			ReturnGroupItem: true,
+		})
+		if err == nil {
+			tables = append(tables, l.getTables(companyId, response)...)
+		}
+		//计算项 计算集
+		response, err = l.svcCtx.ByteMetadataService.ObjectTableSearch(l.ctx, bytelib.ObjectTableSearchRequest{
+			Token:           accessToken,
+			TableTypes:      []string{bytelib.CalculateItem, bytelib.CalculateSet},
+			Module:          bytelib.ModuleCalculateCenter,
+			ReturnGroupItem: true,
+			ExcludeTables:   []int{0},
+		})
+		if err == nil {
+			tables = append(tables, l.getTables(companyId, response)...)
+		}
+	}
+	//获取字段信息
+	fields := l.getFields(tables)
+	//保存数据
+	err = transaction.UseTrans(l.ctx, l.conn.DB(), func(ctx context.Context, conn transaction.Conn) error {
+		err = l.SaveTables(conn, tables)
+		if err != nil {
+			return err
+		}
+		err = l.SaveFields(conn, fields)
+		if err != nil {
+			return err
+		}
+		return nil
+	}, true)
+	return companyIds, err
+}
+
+func (l *SyncTableLogic) getTables(companyId int64, response bytelib.ObjectTableSearchResponse) []*domain.ObjectTable {
+	tables := make([]*domain.ObjectTable, 0)
+	if len(response.List) > 0 {
+		for _, item := range response.List {
+			tables = append(tables, &domain.ObjectTable{
+				Id:            item.Id,
+				TableId:       item.TableId,
+				Name:          item.Name,
+				TableType:     item.TableType,
+				CompanyId:     companyId,
+				ParentId:      item.ParentId,
+				Flag:          item.Flag,
+				Version:       1,
+				IsLocal:       false,
+				RemoteDeleted: 0,
+			})
+		}
+	}
+	return tables
+}
+
+func (l *SyncTableLogic) getFields(list []*domain.ObjectTable) []*domain.ObjectField {
+	fields := make([]*domain.ObjectField, 0)
+	for _, item := range list {
+		accessToken, _ := types.TableAccessToken{CompanyId: item.CompanyId}.GenerateToken()
+		if item.TableId > 0 {
+			response, err := l.svcCtx.ByteMetadataService.TableInfo(l.ctx, &bytelib.TableInfoRequest{
+				Token:   accessToken,
+				TableId: item.TableId,
+			})
+			if err == nil {
+				fields = append(fields, &domain.ObjectField{
+					Id:      int64(item.TableId),
+					Name:    response.Name,
+					Fields:  response.Fields,
+					Version: 1,
+				})
+			}
+		}
+	}
+	return fields
+}
+
+// SaveTables 保存数据
+func (l *SyncTableLogic) SaveTables(conn transaction.Conn, list []*domain.ObjectTable) error {
+	for _, item := range list {
+		objectTable, err := l.svcCtx.ObjectTableRepository.FindOne(l.ctx, conn, item.Id)
+		if err == nil && objectTable.Id > 0 {
+			item.IsLocal = objectTable.IsLocal
+			item.Version = objectTable.Version + 1
+		}
+		_, err = l.svcCtx.ObjectTableRepository.Insert(l.ctx, conn, item)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (l *SyncTableLogic) SaveFields(conn transaction.Conn, list []*domain.ObjectField) error {
+	for _, item := range list {
+		objectTable, err := l.svcCtx.ObjectFieldRepository.FindOne(l.ctx, conn, item.Id)
+		if err == nil && objectTable.Id > 0 {
+			item.Version = objectTable.Version + 1
+		}
+		_, err = l.svcCtx.ObjectFieldRepository.Insert(l.ctx, conn, item)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}

@@ -247,6 +247,9 @@ type SyncTableDataPusher struct {
 	ObjectId  int   `json:"objectId"`
 }
 
+type SyncTableResponse struct {
+}
+
 type AppPageGetRequest struct {
 	Id int64 `path:"id"`
 }

@@ -201,6 +201,18 @@ func (repository *ChartRepository) FindOneByGroup(ctx context.Context, conn tran
 	return repository.ModelToDomainModel(m)
 }
 
+// FindCompanyIds 获取所有的公司ID
+func (repository *ChartRepository) FindCompanyIds(ctx context.Context, conn transaction.Conn) ([]int64, error) {
+	var (
+		err error
+		tx  = conn.DB()
+		m   = new(models.Chart)
+	)
+	list := make([]int64, 0)
+	err = tx.Model(&m).Group("tenant_id").Pluck("tenant_id", &list).Error
+	return list, err
+}
+
 func (repository *ChartRepository) ModelToDomainModel(from *models.Chart) (*domain.Chart, error) {
 	to := &domain.Chart{}
 	err := copier.Copy(to, from)

@@ -29,7 +29,6 @@ func (repository *ObjectFieldRepository) Insert(ctx context.Context, conn transa
 	}
 	dm.Id = m.Id
 	return repository.ModelToDomainModel(m)
-
 }
 
 func (repository *ObjectFieldRepository) Update(ctx context.Context, conn transaction.Conn, dm *domain.ObjectField) (*domain.ObjectField, error) {
@@ -81,7 +80,7 @@ func (repository *ObjectFieldRepository) Delete(ctx context.Context, conn transa
 		m  = &models.ObjectField{Id: dm.Identify().(int64)}
 	)
 	queryFunc := func() (interface{}, error) {
-		tx = tx.Where("id = ?", m.Id).Delete(m)
+		tx = tx.Where("id = ?", m.Id).Unscoped().Delete(m)
 		return m, tx.Error
 	}
 	if _, err := repository.Query(queryFunc, m.CacheKeyFunc()); err != nil {

@@ -81,7 +81,7 @@ func (repository *ObjectTableRepository) Delete(ctx context.Context, conn transa
 		m  = &models.ObjectTable{Id: dm.Identify().(int)}
 	)
 	queryFunc := func() (interface{}, error) {
-		tx = tx.Where("id = ?", m.Id).Delete(m)
+		tx = tx.Where("id = ?", m.Id).Unscoped().Delete(m)
 		return m, tx.Error
 	}
 	if _, err := repository.Query(queryFunc, m.CacheKeyFunc()); err != nil {

@@ -35,6 +35,7 @@ type ChartRepository interface {
 	Find(ctx context.Context, conn transaction.Conn, queryOptions map[string]interface{}) (int64, []*Chart, error)
 	FindOneByGroup(ctx context.Context, conn transaction.Conn, tenantId, pid int64) (*Chart, error)
 	FindByTypeAndName(ctx context.Context, conn transaction.Conn, tenantId int64, t string, name string) (int64, []*Chart, error)
+	FindCompanyIds(ctx context.Context, conn transaction.Conn) ([]int64, error)
 }
 
 /*************** 索引函数 开始****************/

@@ -3,6 +3,7 @@ package domain
 import (
 	"context"
 	"gitlab.fjmaimaimai.com/allied-creation/sumifcc-bchart/cmd/chart-server/interanl/pkg/db/transaction"
+	"gitlab.fjmaimaimai.com/allied-creation/sumifcc-bchart/cmd/chart-server/interanl/pkg/gateway/bytelib"
 	"gorm.io/plugin/soft_delete"
 )
 
@@ -69,14 +70,14 @@ func (m *ObjectNotice) IsDeletedEvent() bool {
 	}
 	//导入模块 取消应用和应用于
 	if m.Event == "table.apply-on" {
-		if m.MetaData.Module&1 == 0 {
+		if m.MetaData.Module&bytelib.ModuleChartTemplate == 0 {
 			return true
 		}
 	}
 	//拆解模块 方案启用、禁用
 	//计算模块 计算项和计算集启用、禁用
 	if m.Event == "table.query.set.update.status" {
-		if m.MetaData.Status&1 == 0 {
+		if m.MetaData.Status&bytelib.ModuleChartTemplate == 0 {
 			return true
 		}
 	}

@@ -26,6 +26,10 @@ service Core {
 	@doc "源数据表-数据"
 	@handler searchTableData
 	post /table/data (SearchTableDataRequest) returns (SearchTableDataResponse)
+
+	@doc "源数据表-初始化同步表(首次使用本地存储执行一次)"
+	@handler syncTable
+	get /table/sync returns (SyncTableResponse)
 }
 
 @server(
@@ -101,4 +105,7 @@ type (
 		CompanyId int64 `json:"companyId,string"` //公司ID
 		ObjectId int `json:"objectId"`
 	}
+	SyncTableResponse {
+
+	}
 )
\ No newline at end of file