Files
kra-new/internal/data/business/activation_pool.go
T

293 lines
9.7 KiB
Go

package business
import (
"context"
"errors"
"sort"
"strings"
"time"
"gorm.io/gorm"
"kra/internal/biz/business"
)
type activationPoolPO struct {
ID, PromoteID uint64
PlatformName, PromoteAccount string
RegisterType int
ClaimFlag int
CreatedAt time.Time
DeletedAt *time.Time
}
func (activationPoolPO) TableName() string { return "sys_platform_promote_active_pool" }
type activationPoolRepo struct {
data Provider
databases PlatformProvider
}
func NewActivationPoolRepo(data Provider, databases PlatformProvider) business.ActivationPoolRepo {
return &activationPoolRepo{data: data, databases: databases}
}
func (r *activationPoolRepo) oa(ctx context.Context) (*gorm.DB, error) {
db, err := r.databases.Database("oa")
if err != nil || db == nil {
return nil, business.ErrActivationSource
}
db = db.WithContext(ctx)
ok, err := businessTableAvailable(db, "sys_platform_promote_active_pool")
if err != nil || !ok {
return nil, business.ErrActivationSource
}
return db, nil
}
func toBizPool(p activationPoolPO) *business.ActivationPoolChannel {
kind := "个人"
if p.RegisterType == 1 {
kind = "公会"
}
return &business.ActivationPoolChannel{ID: p.ID, PromoteID: p.PromoteID, SourceDB: p.PlatformName, Name: p.PromoteAccount, ChannelType: kind, EnteredAt: p.CreatedAt}
}
func (r *activationPoolRepo) PoolChannel(ctx context.Context, id uint64) (*business.ActivationPoolChannel, error) {
db, err := r.oa(ctx)
if err != nil {
return nil, err
}
var p activationPoolPO
if err := db.Where("id=? AND claim_flag=0 AND deleted_at IS NULL", id).Take(&p).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return nil, business.ErrActivationNotFound
}
return nil, business.ErrActivationSource
}
return toBizPool(p), nil
}
func (r *activationPoolRepo) BindingOptions(ctx context.Context, id uint) ([]*business.ActivationBindingOption, error) {
var rows []struct {
SourceDB, Account string
ID uint64
Level int
}
err := r.data.DB().WithContext(ctx).Table("sys_user_bind_platform b").Joins("JOIN sys_users u ON u.id=b.user_id AND u.deleted_at IS NULL").Where("b.user_id=? AND b.status='0' AND u.business_status=0 AND u.enable=1", id).Select("b.source_db,b.platform_busier_id AS id,b.platform_busier_account AS account,u.level").Order("b.source_db,b.platform_busier_id").Scan(&rows).Error
if err != nil {
return nil, business.ErrStorage
}
out := make([]*business.ActivationBindingOption, 0, len(rows))
for _, p := range rows {
db, err := r.databases.Database(p.SourceDB)
if err != nil || db == nil {
return nil, business.ErrPlatformUnavailable
}
var sdk struct{ Level int }
if err := db.WithContext(ctx).Table("tab_promote_business").Where("id=?", p.ID).Select("level").Take(&sdk).Error; err != nil {
return nil, business.ErrPlatformUnavailable
}
p.Level = sdk.Level
out = append(out, &business.ActivationBindingOption{SourceDB: p.SourceDB, Account: p.Account, ID: p.ID, Level: p.Level})
}
return out, nil
}
func (r *activationPoolRepo) Pool(ctx context.Context, f business.ActivationPoolFilter, opts ...business.ListOption) (*business.ActivationPoolPage, error) {
out := &business.ActivationPoolPage{Rows: []*business.ActivationPoolChannel{}, Available: true}
db, err := r.oa(ctx)
if err != nil {
out.Available = false
return out, nil
}
q := db.Model(&activationPoolPO{}).Where("deleted_at IS NULL AND claim_flag=0")
if len(f.Sources) > 0 {
q = q.Where("platform_name IN ?", f.Sources)
}
if f.Name != "" {
q = q.Where("promote_account LIKE ?", "%"+f.Name+"%")
}
if f.ChannelType == "公会" {
q = q.Where("register_type=1")
} else if f.ChannelType == "个人" {
q = q.Where("register_type<>1")
}
var pools []activationPoolPO
if err := q.Order("created_at DESC,id DESC").Find(&pools).Error; err != nil {
return nil, business.ErrActivationSource
}
if len(pools) == 0 {
return out, nil
}
bySource := map[string][]uint64{}
ids := make([]uint64, 0, len(pools))
for _, p := range pools {
bySource[p.PlatformName] = append(bySource[p.PlatformName], p.PromoteID)
ids = append(ids, p.ID)
}
byKey := map[business.ChannelKey]channelPO{}
for source, channelIDs := range bySource {
sdk, err := r.databases.Database(source)
if err != nil || sdk == nil {
out.Available = false
return out, nil
}
sdk = sdk.WithContext(ctx)
for offset := 0; offset < len(channelIDs); offset += 500 {
var channels []channelPO
if err := sdk.Table("tab_promote").Select("id,status,lock_status,last_login_time,create_time").Where("id IN ?", channelIDs[offset:min(offset+500, len(channelIDs))]).Scan(&channels).Error; err != nil {
return nil, business.ErrActivationSource
}
for _, ch := range channels {
ch.SourceDB = source
byKey[business.ChannelKey{SourceDB: source, ID: ch.ID}] = ch
}
}
}
var rejected []struct {
PoolRecordID uint64
RejectReason string
ReviewAt *time.Time
}
if err := db.Table("sys_promote_active_pool_bind b").Select("b.pool_record_id,b.reject_reason,b.review_at").Where("b.pool_record_id IN ? AND b.deleted_at IS NULL AND b.status='rejected'", ids).Where("NOT EXISTS (?)", db.Table("sys_promote_active_pool_bind newer").Select("1").Where("newer.pool_record_id=b.pool_record_id AND newer.deleted_at IS NULL AND newer.id>b.id")).Scan(&rejected).Error; err != nil {
return nil, business.ErrActivationSource
}
rejections := map[uint64]int{}
for i, p := range rejected {
rejections[p.PoolRecordID] = i
}
rows := []*business.ActivationPoolChannel{}
now := time.Now().Unix()
for _, p := range pools {
row := toBizPool(p)
ch, found := byKey[business.ChannelKey{SourceDB: p.PlatformName, ID: p.PromoteID}]
if !found {
continue
}
row.Status = "正常"
if (ch.Status != nil && *ch.Status == -2) || (ch.LockStatus != nil && *ch.LockStatus == -2) {
row.Status = "解除合作"
} else if (ch.Status != nil && *ch.Status == -1) || (ch.LockStatus != nil && *ch.LockStatus < 0) {
row.Status = "锁定"
}
if f.Status != "" && row.Status != f.Status {
continue
}
login := ch.LastLoginTime
if login == nil || *login == 0 {
login = ch.CreateTime
}
if login != nil {
days := (now - *login) / 86400
if days < 0 {
days = 0
}
row.NoLoginDays = &days
}
if i, ok := rejections[p.ID]; ok {
row.Rejected = true
row.RejectReason = rejected[i].RejectReason
row.RejectedAt = rejected[i].ReviewAt
}
rows = append(rows, row)
}
out.Total = int64(len(rows))
o := business.NewListOptions(opts...)
start := min(o.Offset, len(rows))
end := len(rows)
if o.Limit > 0 {
end = min(end, start+o.Limit)
}
out.Rows = rows[start:end]
return out, nil
}
// Event tables are written transactionally by OA in each platform database.
type channelEventPO struct {
ID, PromoteID uint64
PromoteAccount, Kind string
PromoteLevel int
OccurredAt time.Time
OldBusierID, NewBusierID uint64
OldBusierAccount, NewBusierAccount, Operator, Reason string
FromPool, ToPool bool
}
func (channelEventPO) TableName() string { return "tab_promote_business_event" }
func eventSources(sources []string) []string {
if len(sources) == 0 {
return []string{"lx", "hw", "yq"}
}
return sources
}
func (r *activationPoolRepo) Churn(ctx context.Context, f business.ActivationPoolFilter, opts ...business.ListOption) (*business.ActivationChurnPage, error) {
out := &business.ActivationChurnPage{Rows: []*business.ActivationChurn{}, Available: true}
o := business.NewListOptions(opts...)
for _, source := range eventSources(f.Sources) {
db, err := r.databases.Database(source)
if err != nil || db == nil {
out.Available = false
continue
}
db = db.WithContext(ctx)
ok, err := businessTableAvailable(db, "tab_promote_business_event")
if err != nil {
return nil, err
}
if !ok {
out.Available = false
continue
}
q := db.Model(&channelEventPO{}).Where("kind='transfer' AND from_pool=?", false)
if f.Name != "" {
q = q.Where("promote_account LIKE ?", "%"+f.Name+"%")
}
if f.Target != "" {
if strings.Contains("渠道激活池", f.Target) {
q = q.Where("(to_pool=? OR new_busier_account LIKE ?)", true, "%"+f.Target+"%")
} else {
q = q.Where("to_pool=? AND new_busier_account LIKE ?", false, "%"+f.Target+"%")
}
}
if f.Begin != nil {
q = q.Where("occurred_at>=?", *f.Begin)
}
if f.End != nil {
q = q.Where("occurred_at<?", f.End.AddDate(0, 0, 1))
}
var n int64
if err := q.Count(&n).Error; err != nil {
return nil, business.ErrActivationSource
}
out.Total += n
var events []channelEventPO
if err := q.Order("occurred_at DESC,id DESC").Limit(o.Offset + o.Limit).Find(&events).Error; err != nil {
return nil, business.ErrActivationSource
}
for _, e := range events {
to := e.NewBusierAccount
if e.ToPool {
to = "渠道激活池"
}
out.Rows = append(out.Rows, &business.ActivationChurn{ID: e.ID, PromoteID: e.PromoteID, SourceDB: source, Name: e.PromoteAccount, From: e.OldBusierAccount, To: to, Reason: e.Reason, Operator: e.Operator, At: e.OccurredAt, ToPool: e.ToPool})
}
}
if !out.Available {
out.Rows = []*business.ActivationChurn{}
out.Total = 0
return out, nil
}
sort.Slice(out.Rows, func(i, j int) bool {
a, b := out.Rows[i], out.Rows[j]
if !a.At.Equal(b.At) {
return a.At.After(b.At)
}
if a.SourceDB != b.SourceDB {
return a.SourceDB < b.SourceDB
}
return a.ID > b.ID
})
start := min(o.Offset, len(out.Rows))
end := len(out.Rows)
if o.Limit > 0 {
end = min(end, start+o.Limit)
}
out.Rows = out.Rows[start:end]
return out, nil
}