add payment for task

config.toml

-endpoint=":10001"
+endpoint="127.0.0.1:10001"
 metrics_port = 28010
 private_key = "E671C143A110C239B563F702E9F4017CA6B2B2912F675EED9AA4FED684EB30CC"
 
 [redis]
-addr="192.168.1.241:6379"
+addr="127.0.0.1:6379"
 password="123456"
 db=0
+
+[mysql]
+host="127.0.0.1"
+port=3306
+user="root"
+password="12345678"
+database="liuxuzhong"
\ No newline at end of file

config/Config.go

@@ -6,6 +6,14 @@ import (
 	"io/ioutil"
 )
 
+type MysqlConfig struct {
+	Host   string `json:"host" toml:"host"`
+	Port   int    `json:"port" toml:"port"`
+	User   string `json:"user" toml:"user"`
+	Passwd string `json:"passwd" toml:"passwd"`
+	DbName string `json:"database" toml:"database"`
+}
+
 type RedisConfig struct {
 	Addr     string `json:"addr" toml:"addr"`
 	Password string `json:"password" toml:"password"`
@@ -17,6 +25,7 @@ type Config struct {
 	Endpoint   string      `json:"endpoint" toml:"endpoint"`
 	MetricPort int         `json:"metrics_port" toml:"metrics_port"`
 	Redis      RedisConfig `json:"redis" toml:"redis"`
+	DbConfig   MysqlConfig `json:"mysql" toml:"mysql"`
 }
 
 var _cfg *Config = nil

model/db.go

+package model
+
+import (
+	"fmt"
+	"github.com/astaxie/beego/orm"
+	"github.com/odysseus/nodemanager/config"
+)
+
+func DbInit() {
+	// Set up database
+	dbconf := config.GetConfig().DbConfig
+	datasource := fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?charset=utf8", dbconf.User, dbconf.Passwd, dbconf.Host, dbconf.Port, dbconf.DbName)
+	orm.RegisterDriver("mysql", orm.DRMySQL)
+	orm.RegisterDataBase("default", "mysql", datasource)
+	orm.RegisterModel(new(User))
+}

model/user.go

+package model
+
+import (
+	"github.com/astaxie/beego/orm"
+	_ "github.com/go-sql-driver/mysql"
+)
+
+type User struct {
+	ID               uint   `orm:"auto;pk;column(id);description(用户id)"`
+	Name             string `orm:"size(20);null;description(开发者名称)"`
+	Phone            string `orm:"size(15);null;description(电话)"`
+	Mail             string `orm:"size(120);null;description(邮箱)"`
+	ChainAccount     string `orm:"size(42);null;description(区块链账号)"`
+	Type             uint8  `orm:"default(0);description(用户类型：1=自然人 2=企业)"`
+	IsAuthed         uint8  `orm:"default(0);description(是否实名)"`
+	RegisterFrom     uint8  `orm:"default(0);description(注册来源，三方账号)"`
+	RegisterFromDesc string `orm:"size(16);null;description(三方账户描述)"`
+	Username         string `orm:"size(20);default('');description(开发者平台用户名)"`
+	Balance          int64  `orm:"null;description(用户账户余额)"`
+	Level            uint8  `orm:"default(0);description(信用等级，决定免费调用额度)"`
+	CustomID         string `orm:"size(120);null;description(自定义kong的id)"`
+	CreatedTime      uint64 `orm:"default(0);description(创建时间)"`
+	UpdatedTime      uint64 `orm:"default(0);description(更新时间)"`
+	Deleted          uint8  `orm:"default(0);description(逻辑删除)"`
+}
+
+type UserRepository interface {
+	Create(user *User) error
+	GetByUserId(uid int64) (*User, error)
+	Update(user *User) error
+	Delete(user *User) error
+}
+
+type UserRepositoryImpl struct {
+	o orm.Ormer
+}
+
+func NewUserRepository() UserRepository {
+	o := orm.NewOrm()
+	return &UserRepositoryImpl{o}
+}
+
+func (repo *UserRepositoryImpl) Create(user *User) error {
+	_, err := repo.o.Insert(user)
+	return err
+}
+
+func (repo *UserRepositoryImpl) GetByUserId(uid int64) (*User, error) {
+	user := User{ID: uint(uid)}
+	err := repo.o.Read(&user, "Id")
+	if err != nil {
+		return nil, err
+	}
+	return &user, nil
+}
+
+func (repo *UserRepositoryImpl) Update(user *User) error {
+	_, err := repo.o.Update(user)
+	return err
+}
+
+func (repo *UserRepositoryImpl) Delete(user *User) error {
+	_, err := repo.o.Delete(user)
+	return err
+}

server/workerManager.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"github.com/golang/protobuf/proto"
 	lru "github.com/hashicorp/golang-lru"
+	"github.com/odysseus/nodemanager/model"
 	"github.com/odysseus/nodemanager/utils"
 	odysseus "github.com/odysseus/odysseus-protocol/gen/proto/go/base/v1"
 	omanager "github.com/odysseus/odysseus-protocol/gen/proto/go/nodemanager/v1"
@@ -12,6 +13,7 @@ import (
 	log "github.com/sirupsen/logrus"
 	"golang.org/x/crypto/sha3"
 	"math/big"
+	"strconv"
 	"sync"
 	"time"
 )
@@ -46,6 +48,8 @@ type WorkerManager struct {
 	wkRwLock sync.RWMutex
 	quit     chan struct{}
 
+	userRepo model.UserRepository
+
 	node *Node
 }
 
@@ -56,6 +60,7 @@ func NewWorkerManager(rdb *redis.Client, node *Node) *WorkerManager {
 		quit:      make(chan struct{}),
 		rdb:       rdb,
 		node:      node,
+		userRepo:  model.NewUserRepository(),
 	}
 }
 
@@ -278,6 +283,42 @@ func (wm *WorkerManager) manageWorker(worker *Worker) error {
 					} else {
 						log.WithField("taskid", task.TaskId).Debug("post task result")
 					}
+
+					// pay for task.
+					uid, err := strconv.ParseInt(task.TaskUid, 10, 64)
+					if err != nil {
+						log.WithFields(log.Fields{
+							"taskid": task.TaskId,
+							"uid":    task.TaskUid,
+							"error":  err,
+						}).Error("parse task uid failed")
+					}
+					user, err := wm.userRepo.GetByUserId(uid)
+					if err != nil {
+						log.WithFields(log.Fields{
+							"taskid": task.TaskId,
+							"uid":    task.TaskUid,
+							"error":  err,
+						}).Error("get user by uid failed")
+					}
+					if user != nil {
+						fee, _ := strconv.ParseInt(task.TaskFee, 10, 64)
+						user.Balance = user.Balance - fee
+						err = wm.userRepo.Update(user)
+						if err != nil {
+							log.WithFields(log.Fields{
+								"taskid": task.TaskId,
+								"uid":    task.TaskUid,
+								"error":  err,
+							}).Error("update user failed")
+						} else {
+							log.WithFields(log.Fields{
+								"taskid": task.TaskId,
+								"uid":    task.TaskUid,
+								"fee":    fee,
+							}).Info("pay task fee finished")
+						}
+					}
 				}()
 				_ = wm.AddWorker(worker)
 				// todo: post event for task succeed or failed