refactor(jobs): split runtime from task consumers

internal/app/bootstrap.go

@@ -5,8 +5,9 @@ import (
 
 	"github.com/dujiao-next/internal/app/container"
 	"github.com/dujiao-next/internal/app/httpserver"
+	"github.com/dujiao-next/internal/app/jobs"
+	jobconsumer "github.com/dujiao-next/internal/app/jobs/consumer"
 	"github.com/dujiao-next/internal/config"
-	"github.com/dujiao-next/internal/worker"
 )
 
 // BuildRunner 构建服务运行器
@@ -29,8 +30,8 @@ func BuildRunner(cfg *config.Config, mode string) (*Runner, error) {
 
 	// 初始化 Worker 服务
 	if mode == ModeAll || mode == ModeWorker {
-		consumer := worker.NewConsumer(dependencies)
-		workerService, err := worker.NewService(&cfg.Queue, consumer)
+		consumer := jobconsumer.New(dependencies)
+		workerService, err := jobs.NewService(&cfg.Queue, consumer)
 		if err != nil {
 			return nil, err
 		}

internal/app/jobs/consumer/asynq_worker_test.go

@@ -1,4 +1,4 @@
-package worker
+package consumer
 
 import (
 	"context"

internal/app/jobs/consumer/consumer_bot.go

@@ -1,4 +1,4 @@
-package worker
+package consumer
 
 import (
 	"bytes"

internal/app/jobs/consumer/consumer_core.go

@@ -1,4 +1,4 @@
-package worker
+package consumer
 
 import (
 	"github.com/dujiao-next/internal/app/container"
@@ -19,8 +19,8 @@ type orderReader interface {
 	GetByID(id uint) (*orderdomain.Order, error)
 }
 
-// NewConsumer 创建消费者
-func NewConsumer(c *container.Container) *Consumer {
+// New 创建消费者
+func New(c *container.Container) *Consumer {
 	var orders orderReader
 	if c != nil {
 		orders = c.OrderStore

internal/app/jobs/consumer/consumer_maintenance.go

@@ -1,4 +1,4 @@
-package worker
+package consumer
 
 import (
 	"context"

internal/app/jobs/consumer/consumer_order.go

@@ -1,4 +1,4 @@
-package worker
+package consumer
 
 import (
 	"context"

internal/app/jobs/consumer/consumer_upstream.go

@@ -1,4 +1,4 @@
-package worker
+package consumer
 
 import (
 	"context"

internal/app/jobs/consumer/recovery.go

@@ -1,4 +1,4 @@
-package worker
+package consumer
 
 import (
 	"context"

internal/app/jobs/consumer/recovery_test.go

@@ -1,4 +1,4 @@
-package worker
+package consumer
 
 import (
 	"context"

internal/app/jobs/consumer/reseller_accounting_worker_test.go

@@ -1,4 +1,4 @@
-package worker
+package consumer
 
 import (
 	"context"
@@ -41,7 +41,7 @@ func TestResellerConfirmLedgerWorkerMarksDueEntriesAvailable(t *testing.T) {
 		t.Fatalf("seed ledger failed: %v", err)
 	}
 	repo := resellergormstore.New(db)
-	c := NewConsumer(&container.Container{
+	c := New(&container.Container{
 		ResellerAccountingLedger: resellerapplication.NewAccountingLedgerService(repo, 0),
 	})
 	if err := c.handleResellerConfirmLedger(context.Background(), queue.NewResellerConfirmLedgerTask()); err != nil {
@@ -67,7 +67,7 @@ func TestResellerConfirmLedgerTaskType(t *testing.T) {
 }
 
 func TestResellerConfirmLedgerWorkerSkipNilService(t *testing.T) {
-	c := NewConsumer(&container.Container{})
+	c := New(&container.Container{})
 	if err := c.handleResellerConfirmLedger(context.Background(), asynq.NewTask(queue.TaskResellerConfirmLedger, nil)); err != nil {
 		t.Fatalf("nil service should be skipped, got %v", err)
 	}

internal/app/jobs/service.go

@@ -1,9 +1,10 @@
-package worker
+package jobs
 
 import (
 	"context"
 	"errors"
 
+	jobconsumer "github.com/dujiao-next/internal/app/jobs/consumer"
 	"github.com/dujiao-next/internal/config"
 	"github.com/dujiao-next/internal/logger"
 	settingsintegration "github.com/dujiao-next/internal/modules/settings/schema/integration"
@@ -18,11 +19,11 @@ type Service struct {
 	server    *asynq.Server
 	mux       *asynq.ServeMux
 	scheduler *asynq.Scheduler
-	consumer  *Consumer
+	consumer  *jobconsumer.Consumer
 }
 
 // NewService 创建异步队列服务
-func NewService(cfg *config.QueueConfig, consumer *Consumer) (*Service, error) {
+func NewService(cfg *config.QueueConfig, consumer *jobconsumer.Consumer) (*Service, error) {
 	if cfg == nil || !cfg.Enabled {
 		return nil, errors.New("queue disabled")
 	}
@@ -47,7 +48,7 @@ func NewService(cfg *config.QueueConfig, consumer *Consumer) (*Service, error) {
 }
 
 // registerPeriodicTasks 注册所有周期性任务
-func registerPeriodicTasks(scheduler *asynq.Scheduler, consumer *Consumer, cfg *config.QueueConfig) {
+func registerPeriodicTasks(scheduler *asynq.Scheduler, consumer *jobconsumer.Consumer, cfg *config.QueueConfig) {
 	if scheduler == nil || consumer == nil {
 		return
 	}

internal/architecture/complete_migration_guard_test.go

@@ -4,33 +4,18 @@ import (
 	"io/fs"
 	"os"
 	"path/filepath"
-	"sort"
 	"strings"
 	"testing"
 )
 
-// legacyRootGoFileBudgets freeze the remaining horizontal architecture at the
-// start of the complete migration. Every entry is a one-way budget: deleting
-// or moving files is allowed, adding files below these roots is not. The map is
-// removed together with the roots when the migration is complete.
-var legacyRootGoFileBudgets = map[string]int{
-	"internal/worker": 10,
-}
-
-// These are the only production compatibility shims that existed at the
-// migration baseline. They may be deleted, but no replacement shim may be
-// introduced. The final migration gate requires this set to become empty.
-var baselineCompatibilityFiles = map[string]struct{}{}
-
 type packageFileBudget struct {
 	production int
 	total      int
 }
 
-// Existing oversized packages are frozen at their baseline size. All new or
-// already-focused packages use the default limit below. Entries disappear as
-// their packages are split into bounded-context leaf packages.
-var transitionalPackageFileBudgets = map[string]packageFileBudget{
+// Test-only architecture assertions intentionally share one package so they can
+// reuse AST helpers. Production packages have no file-budget exceptions.
+var packageFileBudgetOverrides = map[string]packageFileBudget{
 	"internal/architecture": {production: 0, total: 55},
 }
 
@@ -48,6 +33,7 @@ var completedMigrationPaths = []string{
 	"internal/service",
 	"internal/transport",
 	"internal/wiring",
+	"internal/worker",
 	"internal/models/payment.go",
 	"internal/models/payment_channel.go",
 	"internal/repository/payment_repository.go",
@@ -614,31 +600,6 @@ func TestDatabaseBootstrapIsSeparatedFromPlatformConnection(t *testing.T) {
 	assertDirectoryGoFileBudget(t, migrationRoot, 4)
 }
 
-func TestLegacyHorizontalRootsCanOnlyShrink(t *testing.T) {
-	repositoryRoot := findRepositoryRoot(t)
-
-	paths := make([]string, 0, len(legacyRootGoFileBudgets))
-	for relativePath := range legacyRootGoFileBudgets {
-		paths = append(paths, relativePath)
-	}
-	sort.Strings(paths)
-
-	for _, relativePath := range paths {
-		t.Run(strings.TrimPrefix(relativePath, "internal/"), func(t *testing.T) {
-			absolutePath := filepath.Join(repositoryRoot, filepath.FromSlash(relativePath))
-			count := countGoFilesRecursively(t, absolutePath)
-			if count > legacyRootGoFileBudgets[relativePath] {
-				t.Fatalf(
-					"legacy root %s grew from its migration baseline of %d Go files to %d; move new code into a bounded context",
-					relativePath,
-					legacyRootGoFileBudgets[relativePath],
-					count,
-				)
-			}
-		})
-	}
-}
-
 func TestNoNewCompatibilityOrLegacyProductionFiles(t *testing.T) {
 	repositoryRoot := findRepositoryRoot(t)
 	internalRoot := filepath.Join(repositoryRoot, "internal")
@@ -661,9 +622,7 @@ func TestNoNewCompatibilityOrLegacyProductionFiles(t *testing.T) {
 			return err
 		}
 		relativePath = filepath.ToSlash(relativePath)
-		if _, allowedDuringMigration := baselineCompatibilityFiles[relativePath]; !allowedDuringMigration {
-			t.Errorf("new compatibility or legacy production file is forbidden: %s", relativePath)
-		}
+		t.Errorf("compatibility or legacy production file is forbidden: %s", relativePath)
 		return nil
 	})
 	if err != nil {
@@ -695,8 +654,8 @@ func TestGoPackagesStayWithinFileBudgets(t *testing.T) {
 		relativePath = filepath.ToSlash(relativePath)
 
 		budget := packageFileBudget{production: 12, total: 20}
-		if transitional, ok := transitionalPackageFileBudgets[relativePath]; ok {
-			budget = transitional
+		if override, ok := packageFileBudgetOverrides[relativePath]; ok {
+			budget = override
 		}
 
 		if production > budget.production || total > budget.total {
@@ -716,31 +675,6 @@ func TestGoPackagesStayWithinFileBudgets(t *testing.T) {
 	}
 }
 
-func countGoFilesRecursively(t *testing.T, root string) int {
-	t.Helper()
-	if _, err := os.Stat(root); err != nil {
-		if os.IsNotExist(err) {
-			return 0
-		}
-		t.Fatalf("stat %s: %v", root, err)
-	}
-
-	count := 0
-	err := filepath.WalkDir(root, func(path string, entry fs.DirEntry, walkErr error) error {
-		if walkErr != nil {
-			return walkErr
-		}
-		if !entry.IsDir() && filepath.Ext(path) == ".go" {
-			count++
-		}
-		return nil
-	})
-	if err != nil {
-		t.Fatalf("count Go files below %s: %v", root, err)
-	}
-	return count
-}
-
 func countDirectGoFiles(t *testing.T, directory string) (production int, total int) {
 	t.Helper()
 	entries, err := os.ReadDir(directory)

internal/architecture/worker_consumer_structure_test.go

@@ -8,16 +8,17 @@ import (
 
 func TestWorkerConsumerIsSplitByJobDomain(t *testing.T) {
 	repositoryRoot := findRepositoryRoot(t)
-	workerDirectory := filepath.Join(repositoryRoot, "internal", "worker")
-	legacyPath := filepath.Join(workerDirectory, "asynq_worker.go")
+	jobsDirectory := filepath.Join(repositoryRoot, "internal", "app", "jobs")
+	consumerDirectory := filepath.Join(jobsDirectory, "consumer")
+	legacyPath := filepath.Join(repositoryRoot, "internal", "worker")
 	if _, err := os.Stat(legacyPath); err == nil {
-		t.Fatalf("asynq_worker.go must be replaced by job-domain-focused consumer files")
+		t.Fatalf("legacy worker root must stay deleted")
 	} else if !os.IsNotExist(err) {
-		t.Fatalf("stat asynq_worker.go: %v", err)
+		t.Fatalf("stat legacy worker root: %v", err)
 	}
 
 	expectedOwner := map[string]string{
-		"NewConsumer":                       "consumer_core.go",
+		"New":                               "consumer_core.go",
 		"Register":                          "consumer_core.go",
 		"handleOrderStatusEmail":            "consumer_order.go",
 		"handleOrderAutoFulfill":            "consumer_order.go",
@@ -50,7 +51,7 @@ func TestWorkerConsumerIsSplitByJobDomain(t *testing.T) {
 	actualOwners := make(map[string][]string, len(expectedOwner))
 	consumerTypeOwners := make([]string, 0, 1)
 	for _, file := range files {
-		parsed := parseProductionGoFile(t, filepath.Join(workerDirectory, file))
+		parsed := parseProductionGoFile(t, filepath.Join(consumerDirectory, file))
 		for _, function := range declaredFunctionNames(parsed) {
 			if _, tracked := expectedOwner[function]; tracked {
 				actualOwners[function] = append(actualOwners[function], file)
@@ -72,4 +73,7 @@ func TestWorkerConsumerIsSplitByJobDomain(t *testing.T) {
 	if len(consumerTypeOwners) != 1 || consumerTypeOwners[0] != "consumer_core.go" {
 		t.Errorf("Consumer ownership mismatch: want [consumer_core.go], got %v", consumerTypeOwners)
 	}
+
+	assertFileDeclaresTypes(t, filepath.Join(jobsDirectory, "service.go"), []string{"Service"})
+	assertFileDeclaresFunctions(t, filepath.Join(jobsDirectory, "service.go"), []string{"NewService"})
 }