refactor: 更新项目引用路径从perfect-panel/ppanel-server到perfect-panel/server

feat: 添加版本和构建时间变量 fix: 修正短信队列类型注释错误 style: 清理未使用的代码和测试文件 docs: 更新安装文档中的下载链接 chore: 迁移数据库脚本添加日志和订阅配置
2025-10-13 01:33:03 -07:00
parent 393b42f35a
commit c582087c0f
974 changed files with 23609 additions and 23398 deletions
@@ -0,0 +1,170 @@
+package marketing
+
+import (
+	"context"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/hibiken/asynq"
+	"github.com/perfect-panel/server/internal/model/task"
+	"github.com/perfect-panel/server/internal/model/user"
+	"github.com/perfect-panel/server/internal/svc"
+	"github.com/perfect-panel/server/internal/types"
+	"github.com/perfect-panel/server/pkg/logger"
+	"github.com/perfect-panel/server/pkg/tool"
+	"github.com/perfect-panel/server/pkg/xerr"
+	types2 "github.com/perfect-panel/server/queue/types"
+	"gorm.io/gorm"
+)
+
+type CreateBatchSendEmailTaskLogic struct {
+	logger.Logger
+	ctx    context.Context
+	svcCtx *svc.ServiceContext
+}
+
+// NewCreateBatchSendEmailTaskLogic Create a batch send email task
+func NewCreateBatchSendEmailTaskLogic(ctx context.Context, svcCtx *svc.ServiceContext) *CreateBatchSendEmailTaskLogic {
+	return &CreateBatchSendEmailTaskLogic{
+		Logger: logger.WithContext(ctx),
+		ctx:    ctx,
+		svcCtx: svcCtx,
+	}
+}
+func (l *CreateBatchSendEmailTaskLogic) CreateBatchSendEmailTask(req *types.CreateBatchSendEmailTaskRequest) (err error) {
+	tx := l.svcCtx.DB
+
+	var emails []string
+
+	// 通用查询器（含 user JOIN + 注册时间范围过滤）
+	baseQuery := func() *gorm.DB {
+		query := tx.Model(&user.AuthMethods{}).
+			Select("auth_identifier").
+			Joins("JOIN user ON user.id = user_auth_methods.user_id").
+			Where("auth_type = ?", "email")
+
+		if req.RegisterStartTime != 0 {
+			query = query.Where("user.created_at >= ?", time.UnixMilli(req.RegisterStartTime))
+		}
+		if req.RegisterEndTime != 0 {
+			query = query.Where("user.created_at <= ?", time.UnixMilli(req.RegisterEndTime))
+		}
+		return query
+	}
+
+	var query *gorm.DB
+
+	scope := task.ParseScopeType(req.Scope)
+
+	switch scope {
+	case task.ScopeAll:
+		query = baseQuery()
+
+	case task.ScopeActive:
+		query = baseQuery().
+			Joins("JOIN user_subscribe ON user.id = user_subscribe.user_id").
+			Where("user_subscribe.status IN ?", []int64{1, 2})
+
+	case task.ScopeExpired:
+		query = baseQuery().
+			Joins("JOIN user_subscribe ON user.id = user_subscribe.user_id").
+			Where("user_subscribe.status = ?", 3)
+
+	case task.ScopeNone:
+		query = baseQuery().
+			Joins("LEFT JOIN user_subscribe ON user.id = user_subscribe.user_id").
+			Where("user_subscribe.user_id IS NULL")
+	default:
+
+	}
+	if query != nil {
+		// 执行查询
+		err = query.Pluck("auth_identifier", &emails).Error
+		if err != nil {
+			l.Errorf("[CreateBatchSendEmailTask] Failed to fetch email addresses: %v", err.Error())
+			return xerr.NewErrCode(xerr.DatabaseQueryError)
+		}
+	}
+
+	// 邮箱列表为空，返回错误
+	if len(emails) == 0 && scope != task.ScopeSkip {
+		l.Errorf("[CreateBatchSendEmailTask] No email addresses found for the specified scope")
+		return xerr.NewErrMsg("No email addresses found for the specified scope")
+	}
+
+	// 邮箱地址去重
+	emails = tool.RemoveDuplicateElements(emails...)
+
+	var additionalEmails []string
+	// 追加额外的邮箱地址（不覆盖）
+	if req.Additional != "" {
+		additionalEmails = tool.RemoveDuplicateElements(strings.Split(req.Additional, "\n")...)
+	}
+	if len(additionalEmails) == 0 && scope == task.ScopeSkip {
+		l.Errorf("[CreateBatchSendEmailTask] No additional email addresses provided for skip scope")
+		return xerr.NewErrMsg("No additional email addresses provided for skip scope")
+	}
+
+	scheduledAt := time.Now().Add(10 * time.Second) // 默认延迟10秒执行,防止任务创建和执行时间过于接近
+	if req.Scheduled != 0 {
+		scheduledAt = time.Unix(req.Scheduled, 0)
+		if scheduledAt.Before(time.Now()) {
+			scheduledAt = time.Now()
+		}
+	}
+
+	scopeInfo := task.EmailScope{
+		Type:              scope.Int8(),
+		RegisterStartTime: req.RegisterStartTime,
+		RegisterEndTime:   req.RegisterEndTime,
+		Recipients:        emails,
+		Additional:        additionalEmails,
+		Scheduled:         req.Scheduled,
+		Interval:          req.Interval,
+		Limit:             req.Limit,
+	}
+	scopeBytes, _ := scopeInfo.Marshal()
+
+	taskContent := task.EmailContent{
+		Subject: req.Subject,
+		Content: req.Content,
+	}
+
+	contentBytes, _ := taskContent.Marshal()
+
+	var total uint64
+	if additionalEmails != nil {
+		list := append(emails, additionalEmails...)
+		total = uint64(len(tool.RemoveDuplicateElements(list...)))
+	} else {
+		total = uint64(len(emails))
+	}
+
+	taskInfo := &task.Task{
+		Type:    task.TypeEmail,
+		Scope:   string(scopeBytes),
+		Content: string(contentBytes),
+		Status:  0,
+		Errors:  "",
+		Total:   total,
+		Current: 0,
+	}
+
+	if err = l.svcCtx.DB.Model(&task.Task{}).Create(taskInfo).Error; err != nil {
+		l.Errorf("[CreateBatchSendEmailTask] Failed to create email task: %v", err.Error())
+		return xerr.NewErrCode(xerr.DatabaseInsertError)
+	}
+	// create task
+	l.Infof("[CreateBatchSendEmailTask] Successfully created email task with ID: %d", taskInfo.Id)
+
+	t := asynq.NewTask(types2.ScheduledBatchSendEmail, []byte(strconv.FormatInt(taskInfo.Id, 10)))
+	info, err := l.svcCtx.Queue.EnqueueContext(l.ctx, t, asynq.ProcessAt(scheduledAt))
+	if err != nil {
+		l.Errorf("[CreateBatchSendEmailTask] Failed to enqueue email task: %v", err.Error())
+		return xerr.NewErrCode(xerr.QueueEnqueueError)
+	}
+	l.Infof("[CreateBatchSendEmailTask] Successfully enqueued email task with ID: %s, scheduled at: %s", info.ID, scheduledAt.Format(time.DateTime))
+
+	return nil
+}
@@ -0,0 +1,104 @@
+package marketing
+
+import (
+	"context"
+	"strconv"
+	"time"
+
+	"github.com/hibiken/asynq"
+	"github.com/perfect-panel/server/internal/model/task"
+	"github.com/perfect-panel/server/internal/model/user"
+	"github.com/perfect-panel/server/internal/svc"
+	"github.com/perfect-panel/server/internal/types"
+	"github.com/perfect-panel/server/pkg/logger"
+	"github.com/perfect-panel/server/pkg/xerr"
+	queueType "github.com/perfect-panel/server/queue/types"
+	"github.com/pkg/errors"
+)
+
+type CreateQuotaTaskLogic struct {
+	logger.Logger
+	ctx    context.Context
+	svcCtx *svc.ServiceContext
+}
+
+// NewCreateQuotaTaskLogic Create a quota task
+func NewCreateQuotaTaskLogic(ctx context.Context, svcCtx *svc.ServiceContext) *CreateQuotaTaskLogic {
+	return &CreateQuotaTaskLogic{
+		Logger: logger.WithContext(ctx),
+		ctx:    ctx,
+		svcCtx: svcCtx,
+	}
+}
+
+func (l *CreateQuotaTaskLogic) CreateQuotaTask(req *types.CreateQuotaTaskRequest) error {
+	var subs []*user.Subscribe
+	query := l.svcCtx.DB.WithContext(l.ctx).Model(&user.Subscribe{})
+	if len(req.Subscribers) > 0 {
+		query = query.Where("`subscribe_id` IN ?", req.Subscribers)
+	}
+
+	if req.IsActive != nil && *req.IsActive {
+		query = query.Where("`status` IN ?", []int64{0, 1, 2}) // 0: Pending 1: Active 2: Finished
+	}
+	if req.StartTime != 0 {
+		start := time.UnixMilli(req.StartTime)
+		query = query.Where("`start_time` <= ?", start)
+	}
+	if req.EndTime != 0 {
+		end := time.UnixMilli(req.EndTime)
+		query = query.Where("`expire_time` >= ?", end)
+	}
+
+	if err := query.Find(&subs).Error; err != nil {
+		l.Errorf("[CreateQuotaTask] find subscribers error: %v", err.Error())
+		return errors.Wrapf(xerr.NewErrCode(xerr.DatabaseQueryError), "find subscribers error")
+	}
+	if len(subs) == 0 {
+		return errors.Wrapf(xerr.NewErrMsg("No subscribers found"), "no subscribers found")
+	}
+	var subIds []int64
+	for _, sub := range subs {
+		subIds = append(subIds, sub.Id)
+	}
+
+	scopeInfo := task.QuotaScope{
+		Subscribers: req.Subscribers,
+		IsActive:    req.IsActive,
+		StartTime:   req.StartTime,
+		EndTime:     req.EndTime,
+		Objects:     subIds,
+	}
+	scopeBytes, _ := scopeInfo.Marshal()
+	contentInfo := task.QuotaContent{
+		ResetTraffic: req.ResetTraffic,
+		Days:         req.Days,
+		GiftType:     req.GiftType,
+		GiftValue:    req.GiftValue,
+	}
+	contentBytes, _ := contentInfo.Marshal()
+	// create task
+	newTask := &task.Task{
+		Type:    task.TypeQuota,
+		Status:  0,
+		Scope:   string(scopeBytes),
+		Content: string(contentBytes),
+		Total:   uint64(len(subIds)),
+		Current: 0,
+		Errors:  "",
+	}
+
+	if err := l.svcCtx.DB.WithContext(l.ctx).Model(&task.Task{}).Create(newTask).Error; err != nil {
+		l.Errorf("[CreateQuotaTask] create task error: %v", err.Error())
+		return errors.Wrapf(xerr.NewErrCode(xerr.DatabaseInsertError), "create task error")
+	}
+
+	// enqueue task
+	queueTask := asynq.NewTask(queueType.ForthwithQuotaTask, []byte(strconv.FormatInt(newTask.Id, 10)))
+	if _, err := l.svcCtx.Queue.EnqueueContext(l.ctx, queueTask); err != nil {
+		l.Errorf("[CreateQuotaTask] enqueue task error: %v", err.Error())
+		return errors.Wrapf(xerr.NewErrCode(xerr.QueueEnqueueError), "enqueue task error")
+	}
+	logger.Infof("[CreateQuotaTask] Successfully created task with ID: %d", newTask.Id)
+	return nil
+}
@@ -0,0 +1,89 @@
+package marketing
+
+import (
+	"context"
+	"strings"
+
+	"github.com/perfect-panel/server/internal/model/task"
+	"github.com/perfect-panel/server/internal/svc"
+	"github.com/perfect-panel/server/internal/types"
+	"github.com/perfect-panel/server/pkg/logger"
+	"github.com/perfect-panel/server/pkg/xerr"
+)
+
+type GetBatchSendEmailTaskListLogic struct {
+	logger.Logger
+	ctx    context.Context
+	svcCtx *svc.ServiceContext
+}
+
+// NewGetBatchSendEmailTaskListLogic Get batch send email task list
+func NewGetBatchSendEmailTaskListLogic(ctx context.Context, svcCtx *svc.ServiceContext) *GetBatchSendEmailTaskListLogic {
+	return &GetBatchSendEmailTaskListLogic{
+		Logger: logger.WithContext(ctx),
+		ctx:    ctx,
+		svcCtx: svcCtx,
+	}
+}
+
+func (l *GetBatchSendEmailTaskListLogic) GetBatchSendEmailTaskList(req *types.GetBatchSendEmailTaskListRequest) (resp *types.GetBatchSendEmailTaskListResponse, err error) {
+
+	var tasks []*task.Task
+	tx := l.svcCtx.DB.Model(&task.Task{}).Where("`type` = ?", task.TypeEmail)
+	if req.Status != nil {
+		tx = tx.Where("status = ?", *req.Status)
+	}
+	if req.Scope != nil {
+		tx = tx.Where("scope = ?", req.Scope)
+	}
+	if req.Page == 0 {
+		req.Page = 1
+	}
+	if req.Size == 0 {
+		req.Size = 10
+	}
+	err = tx.Offset((req.Page - 1) * req.Size).Limit(req.Size).Order("created_at DESC").Find(&tasks).Error
+	if err != nil {
+		l.Errorf("failed to get email tasks: %v", err)
+		return nil, xerr.NewErrCode(xerr.DatabaseQueryError)
+	}
+
+	list := make([]types.BatchSendEmailTask, 0)
+
+	for _, t := range tasks {
+		var scopeInfo task.EmailScope
+		if err = scopeInfo.Unmarshal([]byte(t.Scope)); err != nil {
+			l.Errorf("[GetBatchSendEmailTaskList] failed to unmarshal email task scope: %v", err.Error())
+			continue
+		}
+		var contentInfo task.EmailContent
+		if err = contentInfo.Unmarshal([]byte(t.Content)); err != nil {
+			l.Errorf("[GetBatchSendEmailTaskList] failed to unmarshal email task content: %v", err.Error())
+			continue
+		}
+
+		list = append(list, types.BatchSendEmailTask{
+			Id:                t.Id,
+			Subject:           contentInfo.Subject,
+			Content:           contentInfo.Content,
+			Recipients:        strings.Join(scopeInfo.Recipients, "\n"),
+			Scope:             scopeInfo.Type,
+			RegisterStartTime: scopeInfo.RegisterStartTime,
+			RegisterEndTime:   scopeInfo.RegisterEndTime,
+			Additional:        strings.Join(scopeInfo.Additional, "\n"),
+			Scheduled:         scopeInfo.Scheduled,
+			Interval:          scopeInfo.Interval,
+			Limit:             scopeInfo.Limit,
+			Status:            uint8(t.Status),
+			Errors:            t.Errors,
+			Total:             t.Total,
+			Current:           t.Current,
+			CreatedAt:         t.CreatedAt.UnixMilli(),
+			UpdatedAt:         t.UpdatedAt.UnixMilli(),
+		})
+	}
+
+	return &types.GetBatchSendEmailTaskListResponse{
+		List: list,
+	}, nil
+}
@@ -0,0 +1,44 @@
+package marketing
+
+import (
+	"context"
+
+	"github.com/perfect-panel/server/internal/model/task"
+	"github.com/perfect-panel/server/internal/svc"
+	"github.com/perfect-panel/server/internal/types"
+	"github.com/perfect-panel/server/pkg/logger"
+	"github.com/perfect-panel/server/pkg/xerr"
+)
+
+type GetBatchSendEmailTaskStatusLogic struct {
+	logger.Logger
+	ctx    context.Context
+	svcCtx *svc.ServiceContext
+}
+
+// NewGetBatchSendEmailTaskStatusLogic Get batch send email task status
+func NewGetBatchSendEmailTaskStatusLogic(ctx context.Context, svcCtx *svc.ServiceContext) *GetBatchSendEmailTaskStatusLogic {
+	return &GetBatchSendEmailTaskStatusLogic{
+		Logger: logger.WithContext(ctx),
+		ctx:    ctx,
+		svcCtx: svcCtx,
+	}
+}
+
+func (l *GetBatchSendEmailTaskStatusLogic) GetBatchSendEmailTaskStatus(req *types.GetBatchSendEmailTaskStatusRequest) (resp *types.GetBatchSendEmailTaskStatusResponse, err error) {
+	tx := l.svcCtx.DB
+
+	var taskInfo *task.Task
+	err = tx.Model(&task.Task{}).Where("id = ?", req.Id).First(&taskInfo).Error
+	if err != nil {
+		l.Errorf("failed to get email task status, error: %v", err)
+		return nil, xerr.NewErrCode(xerr.DatabaseQueryError)
+	}
+
+	return &types.GetBatchSendEmailTaskStatusResponse{
+		Status:  uint8(taskInfo.Status),
+		Total:   int64(taskInfo.Total),
+		Current: int64(taskInfo.Current),
+		Errors:  taskInfo.Errors,
+	}, nil
+}
@@ -0,0 +1,93 @@
+package marketing
+
+import (
+	"context"
+	"time"
+
+	"github.com/perfect-panel/server/internal/model/task"
+	"github.com/perfect-panel/server/internal/model/user"
+	"github.com/perfect-panel/server/internal/svc"
+	"github.com/perfect-panel/server/internal/types"
+	"github.com/perfect-panel/server/pkg/logger"
+	"github.com/perfect-panel/server/pkg/xerr"
+	"gorm.io/gorm"
+)
+
+type GetPreSendEmailCountLogic struct {
+	logger.Logger
+	ctx    context.Context
+	svcCtx *svc.ServiceContext
+}
+
+// NewGetPreSendEmailCountLogic Get pre-send email count
+func NewGetPreSendEmailCountLogic(ctx context.Context, svcCtx *svc.ServiceContext) *GetPreSendEmailCountLogic {
+	return &GetPreSendEmailCountLogic{
+		Logger: logger.WithContext(ctx),
+		ctx:    ctx,
+		svcCtx: svcCtx,
+	}
+}
+
+func (l *GetPreSendEmailCountLogic) GetPreSendEmailCount(req *types.GetPreSendEmailCountRequest) (resp *types.GetPreSendEmailCountResponse, err error) {
+	tx := l.svcCtx.DB
+	var count int64
+	// 通用查询器（含 user JOIN + 注册时间范围过滤）
+	baseQuery := func() *gorm.DB {
+		query := tx.Model(&user.AuthMethods{}).
+			Select("auth_identifier").
+			Joins("JOIN user ON user.id = user_auth_methods.user_id").
+			Where("auth_type = ?", "email")
+
+		if req.RegisterStartTime != 0 {
+
+			registerStartTime := time.UnixMilli(req.RegisterStartTime)
+
+			query = query.Where("user.created_at >= ?", registerStartTime)
+		}
+		if req.RegisterEndTime != 0 {
+			registerEndTime := time.UnixMilli(req.RegisterEndTime)
+			query = query.Where("user.created_at <= ?", registerEndTime)
+		}
+		return query
+	}
+	var query *gorm.DB
+	scope := task.ParseScopeType(req.Scope)
+
+	switch scope {
+	case task.ScopeAll:
+		query = baseQuery()
+
+	case task.ScopeActive:
+		query = baseQuery().
+			Joins("JOIN user_subscribe ON user.id = user_subscribe.user_id").
+			Where("user_subscribe.status IN ?", []int64{1, 2})
+
+	case task.ScopeExpired:
+		query = baseQuery().
+			Joins("JOIN user_subscribe ON user.id = user_subscribe.user_id").
+			Where("user_subscribe.status = ?", 3)
+
+	case task.ScopeNone:
+		query = baseQuery().
+			Joins("LEFT JOIN user_subscribe ON user.id = user_subscribe.user_id").
+			Where("user_subscribe.user_id IS NULL")
+	case task.ScopeSkip:
+		// Skip scope does not require a count
+		query = nil
+	default:
+		l.Errorf("[CreateBatchSendEmailTask] Invalid scope: %v", req.Scope)
+		return nil, xerr.NewErrMsg("Invalid email scope")
+
+	}
+
+	if query != nil {
+		if err = query.Count(&count).Error; err != nil {
+			l.Errorf("[GetPreSendEmailCount] Count error: %v", err)
+			return nil, xerr.NewErrMsg("Failed to count emails")
+		}
+	}
+
+	return &types.GetPreSendEmailCountResponse{
+		Count: count,
+	}, nil
+}
@@ -0,0 +1,83 @@
+package marketing
+
+import (
+	"context"
+
+	"github.com/perfect-panel/server/internal/model/task"
+	"github.com/perfect-panel/server/internal/svc"
+	"github.com/perfect-panel/server/internal/types"
+	"github.com/perfect-panel/server/pkg/logger"
+)
+
+type QueryQuotaTaskListLogic struct {
+	logger.Logger
+	ctx    context.Context
+	svcCtx *svc.ServiceContext
+}
+
+// NewQueryQuotaTaskListLogic Query quota task list
+func NewQueryQuotaTaskListLogic(ctx context.Context, svcCtx *svc.ServiceContext) *QueryQuotaTaskListLogic {
+	return &QueryQuotaTaskListLogic{
+		Logger: logger.WithContext(ctx),
+		ctx:    ctx,
+		svcCtx: svcCtx,
+	}
+}
+
+func (l *QueryQuotaTaskListLogic) QueryQuotaTaskList(req *types.QueryQuotaTaskListRequest) (resp *types.QueryQuotaTaskListResponse, err error) {
+	var data []*task.Task
+	var count int64
+	query := l.svcCtx.DB.Model(&task.Task{}).Where("`type` = ?", task.TypeQuota)
+	if req.Page == 0 {
+		req.Page = 1
+	}
+	if req.Size == 0 {
+		req.Size = 20
+	}
+
+	if req.Status != nil {
+		query = query.Where("`status` = ?", *req.Status)
+	}
+	err = query.Count(&count).Offset((req.Page - 1) * req.Size).Limit(req.Size).Order("created_at DESC").Find(&data).Error
+	if err != nil {
+		l.Errorf("[QueryQuotaTaskList] failed to get quota tasks: %v", err)
+		return nil, err
+	}
+
+	var list []types.QuotaTask
+	for _, item := range data {
+		var scopeInfo task.QuotaScope
+		if err = scopeInfo.Unmarshal([]byte(item.Scope)); err != nil {
+			l.Errorf("[QueryQuotaTaskList] failed to unmarshal quota task scope: %v", err.Error())
+			continue
+		}
+		var contentInfo task.QuotaContent
+		if err = contentInfo.Unmarshal([]byte(item.Content)); err != nil {
+			l.Errorf("[QueryQuotaTaskList] failed to unmarshal quota task content: %v", err.Error())
+			continue
+		}
+		list = append(list, types.QuotaTask{
+			Id:           item.Id,
+			Subscribers:  scopeInfo.Subscribers,
+			IsActive:     scopeInfo.IsActive,
+			StartTime:    scopeInfo.StartTime,
+			EndTime:      scopeInfo.EndTime,
+			ResetTraffic: contentInfo.ResetTraffic,
+			Days:         contentInfo.Days,
+			GiftType:     contentInfo.GiftType,
+			GiftValue:    contentInfo.GiftValue,
+			Objects:      scopeInfo.Objects,
+			Status:       uint8(item.Status),
+			Total:        int64(item.Total),
+			Current:      int64(item.Current),
+			Errors:       item.Errors,
+			CreatedAt:    item.CreatedAt.UnixMilli(),
+			UpdatedAt:    item.UpdatedAt.UnixMilli(),
+		})
+	}
+
+	return &types.QueryQuotaTaskListResponse{
+		Total: count,
+		List:  list,
+	}, nil
+}
@@ -0,0 +1,55 @@
+package marketing
+
+import (
+	"context"
+	"time"
+
+	"github.com/perfect-panel/server/internal/model/user"
+	"github.com/perfect-panel/server/internal/svc"
+	"github.com/perfect-panel/server/internal/types"
+	"github.com/perfect-panel/server/pkg/logger"
+)
+
+type QueryQuotaTaskPreCountLogic struct {
+	logger.Logger
+	ctx    context.Context
+	svcCtx *svc.ServiceContext
+}
+
+// NewQueryQuotaTaskPreCountLogic Query quota task pre-count
+func NewQueryQuotaTaskPreCountLogic(ctx context.Context, svcCtx *svc.ServiceContext) *QueryQuotaTaskPreCountLogic {
+	return &QueryQuotaTaskPreCountLogic{
+		Logger: logger.WithContext(ctx),
+		ctx:    ctx,
+		svcCtx: svcCtx,
+	}
+}
+
+func (l *QueryQuotaTaskPreCountLogic) QueryQuotaTaskPreCount(req *types.QueryQuotaTaskPreCountRequest) (resp *types.QueryQuotaTaskPreCountResponse, err error) {
+	tx := l.svcCtx.DB.WithContext(l.ctx).Model(&user.Subscribe{})
+	var count int64
+
+	if len(req.Subscribers) > 0 {
+		tx = tx.Where("`subscribe_id` IN ?", req.Subscribers)
+	}
+
+	if req.IsActive != nil && *req.IsActive {
+		tx = tx.Where("`status` IN ?", []int64{0, 1, 2}) // 0: Pending 1: Active 2: Finished
+	}
+	if req.StartTime != 0 {
+		start := time.UnixMilli(req.StartTime)
+		tx = tx.Where("`start_time` <= ?", start)
+	}
+	if req.EndTime != 0 {
+		end := time.UnixMilli(req.EndTime)
+		tx = tx.Where("`expire_time` >= ?", end)
+	}
+	if err = tx.Count(&count).Error; err != nil {
+		l.Errorf("[QueryQuotaTaskPreCount] count error: %v", err.Error())
+		return nil, err
+	}
+
+	return &types.QueryQuotaTaskPreCountResponse{
+		Count: count,
+	}, nil
+}
@@ -0,0 +1,42 @@
+package marketing
+
+import (
+	"context"
+
+	"github.com/perfect-panel/server/internal/model/task"
+	"github.com/perfect-panel/server/internal/svc"
+	"github.com/perfect-panel/server/internal/types"
+	"github.com/perfect-panel/server/pkg/logger"
+	"github.com/perfect-panel/server/pkg/xerr"
+	"github.com/pkg/errors"
+)
+
+type QueryQuotaTaskStatusLogic struct {
+	logger.Logger
+	ctx    context.Context
+	svcCtx *svc.ServiceContext
+}
+
+// NewQueryQuotaTaskStatusLogic Query quota task status
+func NewQueryQuotaTaskStatusLogic(ctx context.Context, svcCtx *svc.ServiceContext) *QueryQuotaTaskStatusLogic {
+	return &QueryQuotaTaskStatusLogic{
+		Logger: logger.WithContext(ctx),
+		ctx:    ctx,
+		svcCtx: svcCtx,
+	}
+}
+
+func (l *QueryQuotaTaskStatusLogic) QueryQuotaTaskStatus(req *types.QueryQuotaTaskStatusRequest) (resp *types.QueryQuotaTaskStatusResponse, err error) {
+	var data *task.Task
+	err = l.svcCtx.DB.Model(&task.Task{}).Where("id = ? AND `type` = ?", req.Id, task.TypeQuota).First(&data).Error
+	if err != nil {
+		l.Errorf("[QueryQuotaTaskStatus] failed to get quota task: %v", err.Error())
+		return nil, errors.Wrapf(xerr.NewErrCode(xerr.DatabaseQueryError), " failed to get quota task: %v", err.Error())
+	}
+	return &types.QueryQuotaTaskStatusResponse{
+		Status:  uint8(data.Status),
+		Current: int64(data.Current),
+		Total:   int64(data.Total),
+		Errors:  data.Errors,
+	}, nil
+}
@@ -0,0 +1,42 @@
+package marketing
+
+import (
+	"context"
+
+	"github.com/perfect-panel/server/internal/model/task"
+	"github.com/perfect-panel/server/internal/svc"
+	"github.com/perfect-panel/server/internal/types"
+	"github.com/perfect-panel/server/pkg/email"
+	"github.com/perfect-panel/server/pkg/logger"
+	"github.com/perfect-panel/server/pkg/xerr"
+)
+
+type StopBatchSendEmailTaskLogic struct {
+	logger.Logger
+	ctx    context.Context
+	svcCtx *svc.ServiceContext
+}
+
+// NewStopBatchSendEmailTaskLogic Stop a batch send email task
+func NewStopBatchSendEmailTaskLogic(ctx context.Context, svcCtx *svc.ServiceContext) *StopBatchSendEmailTaskLogic {
+	return &StopBatchSendEmailTaskLogic{
+		Logger: logger.WithContext(ctx),
+		ctx:    ctx,
+		svcCtx: svcCtx,
+	}
+}
+
+func (l *StopBatchSendEmailTaskLogic) StopBatchSendEmailTask(req *types.StopBatchSendEmailTaskRequest) (err error) {
+	if email.Manager != nil {
+		email.Manager.RemoveWorker(req.Id)
+	} else {
+		logger.Error("[StopBatchSendEmailTaskLogic] email.Manager is nil, cannot stop task")
+	}
+	err = l.svcCtx.DB.Model(&task.Task{}).Where("id = ?", req.Id).Update("status", 2).Error
+
+	if err != nil {
+		l.Errorf("failed to stop email task, error: %v", err)
+		return xerr.NewErrCode(xerr.DatabaseUpdateError)
+	}
+	return
+}