Migration attribute data to credential (#584)

* cherry-pick

* chore: protocol

* chore: user register and update

* chore: credential

* fix: cherry-pick

* chore: user register and update

* feat: register pb

* fix: cherry-pick

* feat: login

* feat: check param

* fix: check param

* fix: protocol

* fix: cherry-pick

* fix: cherry-pick

* fix: old data in attribute to credential

* fix: old data in attribute to credential

start-config.yml

@@ -5,4 +5,5 @@ serviceBinaries:
   admin-rpc: 1
 toolBinaries:
   - check-component
+  - attribute-to-credential
 maxFileDescriptors: 10000

tools/attribute-to-credential/main.go

@@ -0,0 +1,153 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"github.com/openimsdk/chat/internal/rpc/chat"
+	"github.com/openimsdk/chat/pkg/common/cmd"
+	"github.com/openimsdk/chat/pkg/common/config"
+	"github.com/openimsdk/chat/pkg/common/constant"
+	table "github.com/openimsdk/chat/pkg/common/db/table/chat"
+	"github.com/openimsdk/chat/tools/dataversion"
+	"github.com/openimsdk/protocol/sdkws"
+	"github.com/openimsdk/tools/db/mongoutil"
+	"github.com/openimsdk/tools/system/program"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+	"path/filepath"
+)
+
+const (
+	credentialKey     = "credential"
+	credentialVersion = 1
+
+	attributeCollection  = "attribute"
+	credentialCollection = "credential"
+	pageNum              = 1000
+)
+
+func initConfig(configDir string) (*config.Mongo, error) {
+	var (
+		mongoConfig = &config.Mongo{}
+	)
+	err := config.LoadConfig(filepath.Join(configDir, cmd.MongodbConfigFileName), cmd.ConfigEnvPrefixMap[cmd.MongodbConfigFileName], mongoConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	return mongoConfig, nil
+}
+
+func pageGetAttribute(ctx context.Context, coll *mongo.Collection, pagination *sdkws.RequestPagination) (int64, []*table.Attribute, error) {
+	return mongoutil.FindPage[*table.Attribute](ctx, coll, bson.M{}, pagination)
+}
+
+func doAttributeToCredential() error {
+	var index int
+	var configDir string
+	flag.IntVar(&index, "i", 0, "Index number")
+	defaultConfigDir := filepath.Join("..", "..", "..", "..", "..", "config")
+	flag.StringVar(&configDir, "c", defaultConfigDir, "Configuration dir")
+	flag.Parse()
+
+	fmt.Printf("Index: %d, Config Path: %s\n", index, configDir)
+
+	mongoConfig, err := initConfig(configDir)
+	if err != nil {
+		return err
+	}
+
+	ctx := context.Background()
+
+	mgocli, err := mongoutil.NewMongoDB(ctx, mongoConfig.Build())
+	if err != nil {
+		return err
+	}
+
+	versionColl := mgocli.GetDB().Collection(dataversion.Collection)
+	converted, err := dataversion.CheckVersion(versionColl, credentialKey, credentialVersion)
+	if err != nil {
+		return err
+	}
+	if converted {
+		fmt.Println("[credential] credential data has been converted")
+		return nil
+	}
+
+	attrColl := mgocli.GetDB().Collection(attributeCollection)
+	credColl := mgocli.GetDB().Collection(credentialCollection)
+
+	pagination := &sdkws.RequestPagination{
+		PageNumber: 1,
+		ShowNumber: pageNum,
+	}
+	tx := mgocli.GetTx()
+	if err = tx.Transaction(ctx, func(ctx context.Context) error {
+		for {
+			total, attrs, err := pageGetAttribute(ctx, attrColl, pagination)
+			if err != nil {
+				return err
+			}
+			credentials := make([]*table.Credential, 0, pageNum*3)
+			for _, attr := range attrs {
+				if attr.Email != "" {
+					credentials = append(credentials, &table.Credential{
+						UserID:      attr.UserID,
+						Account:     attr.Email,
+						Type:        constant.CredentialEmail,
+						AllowChange: true,
+					})
+				}
+				if attr.Account != "" {
+					credentials = append(credentials, &table.Credential{
+						UserID:      attr.UserID,
+						Account:     attr.Account,
+						Type:        constant.CredentialAccount,
+						AllowChange: true,
+					})
+				}
+				if attr.PhoneNumber != "" && attr.AreaCode != "" {
+					credentials = append(credentials, &table.Credential{
+						UserID:      attr.UserID,
+						Account:     chat.BuildCredentialPhone(attr.AreaCode, attr.PhoneNumber),
+						Type:        constant.CredentialPhone,
+						AllowChange: true,
+					})
+				}
+
+			}
+			for _, credential := range credentials {
+				err = mongoutil.UpdateOne(ctx, credColl, bson.M{
+					"user_id": credential.UserID,
+					"type":    credential.Type,
+				}, bson.M{
+					"$set": credential,
+				}, false, options.Update().SetUpsert(true))
+				if err != nil {
+					return err
+				}
+			}
+
+			pagination.PageNumber++
+			if total < pageNum {
+				break
+			}
+		}
+		return nil
+	}); err != nil {
+		return err
+	}
+	if err := dataversion.SetVersion(versionColl, credentialKey, credentialVersion); err != nil {
+		return fmt.Errorf("set mongodb credential version %w", err)
+	}
+	fmt.Println("[credential] update old data to credential success")
+	return nil
+}
+
+func main() {
+	if err := doAttributeToCredential(); err != nil {
+		program.ExitWithError(err)
+	}
+}

tools/dataversion/data_version.go

@@ -0,0 +1,50 @@
+package dataversion
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"github.com/openimsdk/tools/db/mongoutil"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+	"strconv"
+	"time"
+)
+
+const (
+	Collection = "data_version"
+)
+
+func CheckVersion(coll *mongo.Collection, key string, currentVersion int) (converted bool, err error) {
+	type VersionTable struct {
+		Key   string `bson:"key"`
+		Value string `bson:"value"`
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
+	defer cancel()
+	res, err := mongoutil.FindOne[VersionTable](ctx, coll, bson.M{"key": key})
+	if err == nil {
+		ver, err := strconv.Atoi(res.Value)
+		if err != nil {
+			return false, fmt.Errorf("version %s parse error %w", res.Value, err)
+		}
+		if ver >= currentVersion {
+			return true, nil
+		}
+		return false, nil
+	} else if errors.Is(err, mongo.ErrNoDocuments) {
+		return false, nil
+	} else {
+		return false, err
+	}
+}
+
+func SetVersion(coll *mongo.Collection, key string, version int) error {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
+	defer cancel()
+	option := options.Update().SetUpsert(true)
+	filter := bson.M{"key": key, "value": strconv.Itoa(version)}
+	update := bson.M{"$set": bson.M{"key": key, "value": strconv.Itoa(version)}}
+	return mongoutil.UpdateOne(ctx, coll, filter, update, false, option)
+}