6843c7efac
后端：
1.阶段 5 task 服务边界落地
- 新增 cmd/task 与 services/task/{dao,rpc,sv}，承载 task zrpc、tasks 表迁移和 task outbox 消费边界
- 新增 gateway/client/task、shared/contracts/task 和 task port，gateway /api/v1/task/* 切到 task zrpc client
- 将 task.urgency.promote.requested handler / relay / retry loop 迁入 cmd/task，单体 worker 不再消费 task outbox
- 保留单体 Agent 残留 task 查询的 publish-only 写入能力，避免迁移期 task 事件丢失
- active-scheduler task facts / due job scanner 切到 task RPC，并移除启动期 tasks 表依赖检查
- 更新阶段 5 文档，记录 task 切流点、旧实现保留、跨域 DB 依赖缩减和下一轮建议
- 补充 task rpc 示例配置
84 lines
2.5 KiB
Go
84 lines
2.5 KiB
Go
package main
|
||
|
||
import (
|
||
"context"
|
||
"log"
|
||
"os"
|
||
"os/signal"
|
||
"syscall"
|
||
|
||
"github.com/LoveLosita/smartflow/backend/bootstrap"
|
||
rootdao "github.com/LoveLosita/smartflow/backend/dao"
|
||
kafkabus "github.com/LoveLosita/smartflow/backend/infra/kafka"
|
||
outboxinfra "github.com/LoveLosita/smartflow/backend/infra/outbox"
|
||
rootmiddleware "github.com/LoveLosita/smartflow/backend/middleware"
|
||
taskdao "github.com/LoveLosita/smartflow/backend/services/task/dao"
|
||
taskrpc "github.com/LoveLosita/smartflow/backend/services/task/rpc"
|
||
tasksv "github.com/LoveLosita/smartflow/backend/services/task/sv"
|
||
"github.com/spf13/viper"
|
||
)
|
||
|
||
func main() {
|
||
if err := bootstrap.LoadConfig(); err != nil {
|
||
log.Fatalf("failed to load config: %v", err)
|
||
}
|
||
|
||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
||
defer stop()
|
||
|
||
db, err := taskdao.OpenDBFromConfig()
|
||
if err != nil {
|
||
log.Fatalf("failed to connect task database: %v", err)
|
||
}
|
||
redisClient, err := taskdao.OpenRedisFromConfig()
|
||
if err != nil {
|
||
log.Fatalf("failed to connect task redis: %v", err)
|
||
}
|
||
defer redisClient.Close()
|
||
|
||
cacheRepo := rootdao.NewCacheDAO(redisClient)
|
||
if err := db.Use(rootmiddleware.NewGormCachePlugin(cacheRepo)); err != nil {
|
||
log.Fatalf("failed to initialize task cache deleter: %v", err)
|
||
}
|
||
|
||
taskRepo := taskdao.NewTaskDAO(db)
|
||
outboxRepo := outboxinfra.NewRepository(db)
|
||
eventBus, err := outboxinfra.NewEventBus(outboxRepo, kafkabus.LoadConfig())
|
||
if err != nil {
|
||
log.Fatalf("failed to initialize task outbox bus: %v", err)
|
||
}
|
||
|
||
svc := tasksv.NewTaskService(taskRepo, cacheRepo, eventBus)
|
||
// 迁移期 task 服务仍 best-effort 维护 active-scheduler due job，后续改成 RPC/事件后再移除该跨域 DAO。
|
||
svc.SetActiveScheduleDAO(rootdao.NewActiveScheduleDAO(db))
|
||
|
||
if eventBus != nil {
|
||
if err := tasksv.RegisterTaskUrgencyPromoteHandler(eventBus, outboxRepo, taskRepo); err != nil {
|
||
log.Fatalf("failed to register task outbox handler: %v", err)
|
||
}
|
||
eventBus.Start(ctx)
|
||
defer eventBus.Close()
|
||
log.Println("Task outbox consumer started")
|
||
} else {
|
||
log.Println("Task outbox consumer is disabled")
|
||
}
|
||
|
||
server, listenOn, err := taskrpc.NewServer(taskrpc.ServerOptions{
|
||
ListenOn: viper.GetString("task.rpc.listenOn"),
|
||
Timeout: viper.GetDuration("task.rpc.timeout"),
|
||
Service: svc,
|
||
})
|
||
if err != nil {
|
||
log.Fatalf("failed to build task zrpc server: %v", err)
|
||
}
|
||
defer server.Stop()
|
||
|
||
go func() {
|
||
log.Printf("task zrpc service starting on %s", listenOn)
|
||
server.Start()
|
||
}()
|
||
|
||
<-ctx.Done()
|
||
log.Println("task service stopping")
|
||
}
|