openbilibili/app/tool/saga/dao/redis.go

package dao

import (
	"context"
	"encoding/json"
	"fmt"

	"go-common/app/tool/saga/model"
	"go-common/library/cache/redis"

	"github.com/pkg/errors"
)

func mergeTaskKey(taskType int) string {
	return fmt.Sprintf("saga_task_%d", taskType)
}

func mrIIDKey(mrIID int) string {
	return fmt.Sprintf("saga_mrIID_%d", mrIID)
}

func (d *Dao) pingRedis(c context.Context) (err error) {
	conn := d.redis.Get(c)
	_, err = conn.Do("SET", "PING", "PONG")
	conn.Close()
	return
}

// AddMRIID ...
func (d *Dao) AddMRIID(c context.Context, mrIID int, expire int) (err error) {
	var (
		key  = mrIIDKey(mrIID)
		conn = d.redis.Get(c)
	)
	defer conn.Close()
	if err = conn.Send("SET", key, mrIID); err != nil {
		return
	}
	if err = conn.Flush(); err != nil {
		return
	}
	if _, err = conn.Receive(); err != nil {
		return
	}
	if _, err = conn.Do("EXPIRE", key, expire); err != nil {
		return
	}
	return
}

// ExistMRIID ...
func (d *Dao) ExistMRIID(c context.Context, mrIID int) (ok bool, err error) {
	var (
		key  = mrIIDKey(mrIID)
		conn = d.redis.Get(c)
	)
	defer conn.Close()
	if _, err = redis.Int(conn.Do("GET", key)); err != nil {
		if err == redis.ErrNil {
			err = nil
		}
		return false, err
	}
	return true, nil
}

// DeleteMRIID ...
func (d *Dao) DeleteMRIID(c context.Context, mrIID int) (err error) {
	var (
		key  = mrIIDKey(mrIID)
		conn = d.redis.Get(c)
	)
	defer conn.Close()
	if err = conn.Send("DEL", key); err != nil {
		return
	}
	if err = conn.Flush(); err != nil {
		return
	}
	if _, err = conn.Receive(); err != nil {
		return
	}
	return
}

// PushMergeTask ...
func (d *Dao) PushMergeTask(c context.Context, taskType int, taskInfo *model.TaskInfo) (err error) {
	var (
		key  = mergeTaskKey(taskType)
		conn = d.redis.Get(c)
		bs   []byte
	)
	defer conn.Close()
	if bs, err = json.Marshal(taskInfo); err != nil {
		err = errors.WithStack(err)
		return
	}
	if err = conn.Send("LPUSH", key, bs); err != nil {
		return
	}
	if err = conn.Flush(); err != nil {
		return
	}
	if _, err = conn.Receive(); err != nil {
		return
	}
	return
}

// DeleteMergeTask ...
func (d *Dao) DeleteMergeTask(c context.Context, taskType int, taskInfo *model.TaskInfo) (err error) {
	var (
		key  = mergeTaskKey(taskType)
		conn = d.redis.Get(c)
		bs   []byte
	)
	defer conn.Close()
	if bs, err = json.Marshal(taskInfo); err != nil {
		err = errors.WithStack(err)
		return
	}
	if err = conn.Send("LREM", key, 0, bs); err != nil {
		return
	}
	if err = conn.Flush(); err != nil {
		return
	}
	if _, err = conn.Receive(); err != nil {
		return
	}
	return
}

// MergeTasks ...
func (d *Dao) MergeTasks(c context.Context, taskType int) (count int, taskInfos []*model.TaskInfo, err error) {
	var (
		key    = mergeTaskKey(taskType)
		values [][]byte
		conn   = d.redis.Get(c)
	)
	defer conn.Close()
	if count, err = redis.Int(conn.Do("LLEN", key)); err != nil {
		return
	}
	if values, err = redis.ByteSlices(conn.Do("LRANGE", key, 0, -1)); err != nil {
		if err == redis.ErrNil {
			err = nil
		}
		return
	}
	taskInfos = make([]*model.TaskInfo, 0, count)
	for _, value := range values {
		taskInfo := &model.TaskInfo{}
		if err = json.Unmarshal(value, &taskInfo); err != nil {
			err = errors.WithStack(err)
			return
		}
		taskInfos = append(taskInfos, taskInfo)
		//taskInfos = append([]*model.TaskInfo{taskInfo}, taskInfos...)
	}
	return
}

func mergeInfoKey(projID int, branch string) string {
	return fmt.Sprintf("saga_mergeInfo_%d_%s", projID, branch)
}

func pathOwnerKey(projID int, branch string, path string) string {
	return fmt.Sprintf("saga_PathOwner_%d_%s_%s", projID, branch, path)
}

func pathReviewerKey(projID int, branch string, path string) string {
	return fmt.Sprintf("saga_PathReviewer_%d_%s_%s", projID, branch, path)
}

func authInfoKey(projID int, mrIID int) string {
	return fmt.Sprintf("saga_auth_%d_%d", projID, mrIID)
}

// SetMergeInfo ...
func (d *Dao) SetMergeInfo(c context.Context, projID int, branch string, mergeInfo *model.MergeInfo) (err error) {
	var (
		key  = mergeInfoKey(projID, branch)
		conn = d.redis.Get(c)
		bs   []byte
	)
	defer conn.Close()
	if bs, err = json.Marshal(mergeInfo); err != nil {
		err = errors.WithStack(err)
		return
	}
	if err = conn.Send("SET", key, bs); err != nil {
		return
	}
	if err = conn.Flush(); err != nil {
		return
	}
	if _, err = conn.Receive(); err != nil {
		return
	}
	return
}

// MergeInfo ...
func (d *Dao) MergeInfo(c context.Context, projID int, branch string) (ok bool, mergeInfo *model.MergeInfo, err error) {
	var (
		key   = mergeInfoKey(projID, branch)
		value []byte
		conn  = d.redis.Get(c)
	)
	defer conn.Close()
	if value, err = redis.Bytes(conn.Do("GET", key)); err != nil {
		if err == redis.ErrNil {
			err = nil
		}
		return
	}
	mergeInfo = &model.MergeInfo{}
	if err = json.Unmarshal(value, &mergeInfo); err != nil {
		err = errors.WithStack(err)
		return
	}
	ok = true
	return
}

// DeleteMergeInfo ...
func (d *Dao) DeleteMergeInfo(c context.Context, projID int, branch string) (err error) {
	var (
		key  = mergeInfoKey(projID, branch)
		conn = d.redis.Get(c)
	)
	defer conn.Close()
	if err = conn.Send("DEL", key); err != nil {
		return
	}
	if err = conn.Flush(); err != nil {
		return
	}
	if _, err = conn.Receive(); err != nil {
		return
	}
	return
}

// SetPathOwner ...
func (d *Dao) SetPathOwner(c context.Context, projID int, branch string, path string, owners []string) (err error) {
	var (
		key  = pathOwnerKey(projID, branch, path)
		conn = d.redis.Get(c)
		bs   []byte
	)
	defer conn.Close()
	if bs, err = json.Marshal(owners); err != nil {
		err = errors.WithStack(err)
		return
	}
	if err = conn.Send("SET", key, bs); err != nil {
		return
	}
	if err = conn.Flush(); err != nil {
		return
	}
	if _, err = conn.Receive(); err != nil {
		return
	}
	return
}

// PathOwner ...
func (d *Dao) PathOwner(c context.Context, projID int, branch string, path string) (owners []string, err error) {
	var (
		key  = pathOwnerKey(projID, branch, path)
		conn = d.redis.Get(c)
		bs   []byte
	)
	defer conn.Close()
	if bs, err = redis.Bytes(conn.Do("GET", key)); err != nil {
		if err == redis.ErrNil {
			err = nil
		}
		return
	}
	if err = json.Unmarshal(bs, &owners); err != nil {
		err = errors.WithStack(err)
		return
	}
	return
}

// SetPathReviewer ...
func (d *Dao) SetPathReviewer(c context.Context, projID int, branch string, path string, reviewers []string) (err error) {
	var (
		key  = pathReviewerKey(projID, branch, path)
		conn = d.redis.Get(c)
		bs   []byte
	)
	defer conn.Close()
	if bs, err = json.Marshal(reviewers); err != nil {
		return errors.WithStack(err)
	}
	if err = conn.Send("SET", key, bs); err != nil {
		return
	}
	if err = conn.Flush(); err != nil {
		return
	}
	if _, err = conn.Receive(); err != nil {
		return
	}
	return
}

// PathReviewer ...
func (d *Dao) PathReviewer(c context.Context, projID int, branch string, path string) (reviewers []string, err error) {
	var (
		key  = pathReviewerKey(projID, branch, path)
		conn = d.redis.Get(c)
		bs   []byte
	)
	defer conn.Close()
	if bs, err = redis.Bytes(conn.Do("GET", key)); err != nil {
		if err == redis.ErrNil {
			err = nil
		}
		return
	}
	if err = json.Unmarshal(bs, &reviewers); err != nil {
		err = errors.WithStack(err)
		return
	}
	return
}

// pathAuthKey ...
func pathAuthKey(projID int, branch string, path string) string {
	return fmt.Sprintf("saga_path_auth_%d_%s_%s", projID, branch, path)
}

// PathAuthR ...
func (d *Dao) PathAuthR(c context.Context, projID int, branch string, path string) (authUsers *model.AuthUsers, err error) {
	var (
		key  = pathAuthKey(projID, branch, path)
		conn = d.redis.Get(c)
		bs   []byte
	)
	defer conn.Close()
	if bs, err = redis.Bytes(conn.Do("GET", key)); err != nil {
		if err == redis.ErrNil {
			err = nil
		}
		return
	}
	authUsers = new(model.AuthUsers)
	if err = json.Unmarshal(bs, authUsers); err != nil {
		err = errors.WithStack(err)
		return
	}
	return
}

// SetPathAuthR ...
func (d *Dao) SetPathAuthR(c context.Context, projID int, branch string, path string, authUsers *model.AuthUsers) (err error) {
	var (
		key  = pathAuthKey(projID, branch, path)
		conn = d.redis.Get(c)
		bs   []byte
	)
	defer conn.Close()
	if bs, err = json.Marshal(authUsers); err != nil {
		return errors.WithStack(err)
	}
	if err = conn.Send("SET", key, bs); err != nil {
		return
	}
	if err = conn.Flush(); err != nil {
		return
	}
	if _, err = conn.Receive(); err != nil {
		return
	}
	return
}

// DeletePathAuthR ...
func (d *Dao) DeletePathAuthR(c context.Context, projID int, branch string, path string) (err error) {
	var (
		key  = pathAuthKey(projID, branch, path)
		conn = d.redis.Get(c)
	)
	defer conn.Close()
	if err = conn.Send("DEL", key); err != nil {
		return
	}
	if err = conn.Flush(); err != nil {
		return
	}
	if _, err = conn.Receive(); err != nil {
		return
	}
	return
}

// SetReportStatus ...
func (d *Dao) SetReportStatus(c context.Context, projID int, mrIID int, result bool) (err error) {
	var (
		key  = authInfoKey(projID, mrIID)
		conn = d.redis.Get(c)
		bs   []byte
	)
	defer conn.Close()
	if bs, err = json.Marshal(result); err != nil {
		return errors.WithStack(err)
	}
	if err = conn.Send("SET", key, bs); err != nil {
		return
	}
	if err = conn.Flush(); err != nil {
		return
	}
	if _, err = conn.Receive(); err != nil {
		return
	}
	return
}

// ReportStatus ...
func (d *Dao) ReportStatus(c context.Context, projID int, mrIID int) (result bool, err error) {
	var (
		key   = authInfoKey(projID, mrIID)
		value []byte
		conn  = d.redis.Get(c)
	)
	defer conn.Close()
	if value, err = redis.Bytes(conn.Do("GET", key)); err != nil {
		if err == redis.ErrNil {
			err = nil
		}
		return
	}
	if err = json.Unmarshal(value, &result); err != nil {
		err = errors.WithStack(err)
		return
	}
	return
}

// DeleteReportStatus ...
func (d *Dao) DeleteReportStatus(c context.Context, projID int, mrIID int) (err error) {
	var (
		key  = authInfoKey(projID, mrIID)
		conn = d.redis.Get(c)
	)
	defer conn.Close()
	if err = conn.Send("DEL", key); err != nil {
		return
	}
	if err = conn.Flush(); err != nil {
		return
	}
	if _, err = conn.Receive(); err != nil {
		return
	}
	return
}