姚章浩
305 lines
9.6 KiB
Go
305 lines
9.6 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"log"
|
|
"net/http"
|
|
"os"
|
|
"os/signal"
|
|
"strings"
|
|
"syscall"
|
|
"time"
|
|
|
|
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/agentbinding"
|
|
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/audit"
|
|
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/channels/dingtalk"
|
|
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/coverage"
|
|
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/directory"
|
|
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/faqtable"
|
|
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/feedback"
|
|
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/httpapi"
|
|
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/qa"
|
|
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/session"
|
|
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/store/postgres"
|
|
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/weknora"
|
|
)
|
|
|
|
var (
|
|
version = "dev"
|
|
commitSHA = "unknown"
|
|
buildTime = "unknown"
|
|
)
|
|
|
|
func runtimeBuildInfo() httpapi.BuildInfo {
|
|
return httpapi.BuildInfo{
|
|
Version: version,
|
|
CommitSHA: commitSHA,
|
|
BuildTime: buildTime,
|
|
}
|
|
}
|
|
|
|
func main() {
|
|
ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
|
defer cancel()
|
|
storeCtx, storeCancel := context.WithTimeout(ctx, 10*time.Second)
|
|
defer storeCancel()
|
|
store, err := postgres.Open(storeCtx, os.Getenv("DATABASE_URL"))
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
defer store.Close()
|
|
client, err := weknora.NewClient(os.Getenv("WEKNORA_BASE_URL"), os.Getenv("WEKNORA_API_KEY"), 120*time.Second)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
bindings, err := agentbinding.New(os.Getenv("DOBOTAI_AGENT_BINDINGS"))
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
groupPolicy := session.GroupPolicy(strings.TrimSpace(os.Getenv("DOBOTAI_GROUP_SESSION_POLICY")))
|
|
if groupPolicy == "" {
|
|
groupPolicy = session.GroupShared
|
|
}
|
|
if groupPolicy != session.GroupShared && groupPolicy != session.GroupPerUser {
|
|
log.Fatal("DOBOTAI_GROUP_SESSION_POLICY must be shared or per_user")
|
|
}
|
|
coverageService, err := coverage.NewService(store)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
qaService, err := qa.New(
|
|
store,
|
|
bindings,
|
|
client,
|
|
groupPolicy,
|
|
qa.WithSucceededTurnObserver(coverageService, func(err error) {
|
|
log.Printf("coverage persistence failed: %v", err)
|
|
}),
|
|
)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
address := strings.TrimSpace(os.Getenv("DOBOTAI_LISTEN_ADDR"))
|
|
if address == "" {
|
|
address = "127.0.0.1:8090"
|
|
}
|
|
auditOptions, err := buildAuditOptions(store, os.Getenv("DOBOTAI_TENANT_KEY"), os.Getenv("DOBOTAI_ADMIN_TOKEN"), address)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
auditOptions.BuildInfo = runtimeBuildInfo()
|
|
api, err := httpapi.NewWithOptions(qaService, os.Getenv("DOBOTAI_INGRESS_TOKEN"), auditOptions)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
|
|
dingCfg := dingtalk.StreamConfig{
|
|
ClientID: os.Getenv("DINGTALK_CLIENT_ID"),
|
|
ClientSecret: os.Getenv("DINGTALK_CLIENT_SECRET"),
|
|
TenantKey: os.Getenv("DOBOTAI_TENANT_KEY"),
|
|
}
|
|
if err := dingCfg.Validate(); err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
faqCfg, err := readFAQConfig()
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
var faqWorkflowDone chan struct{}
|
|
var faqRecordLinks *faqtable.RecordLinkBuilder
|
|
if faqCfg.BaseID != "" {
|
|
faqRecordLinks, err = faqtable.NewRecordLinkBuilder(faqCfg.BaseID, faqCfg.SheetID, faqCfg.ViewID)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
faqRepo, err := postgres.NewFAQExportRepository(store)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
faqDispatcher, err := buildFAQDispatcher(faqRepo, dingCfg.ClientID, dingCfg.ClientSecret, faqCfg)
|
|
if err != nil {
|
|
log.Fatalf("DingTalk FAQ table dispatcher init failed: %v", err)
|
|
}
|
|
faqWorkflowDone = make(chan struct{})
|
|
go func() {
|
|
defer close(faqWorkflowDone)
|
|
runFAQPoller(ctx, faqDispatcher, 5*time.Second)
|
|
}()
|
|
log.Printf("DingTalk FAQ table dispatcher started")
|
|
}
|
|
|
|
supportDispatcher, err := buildSupportDispatcher(
|
|
store,
|
|
dingCfg.ClientID,
|
|
dingCfg.ClientSecret,
|
|
os.Getenv("DINGTALK_SUPPORT_GROUP_OPEN_CONVERSATION_ID"),
|
|
faqRecordLinks,
|
|
)
|
|
if err != nil {
|
|
log.Fatalf("DingTalk support notification init failed: %v", err)
|
|
}
|
|
if supportDispatcher != nil {
|
|
go runSupportDispatcher(ctx, supportDispatcher, 5*time.Second)
|
|
log.Printf("DingTalk support notification dispatcher started")
|
|
}
|
|
// Question table dispatcher (optional, fail-closed).
|
|
qtCfg, err := readQuestionTableConfig()
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
var questionTableDone chan struct{}
|
|
if qtCfg.BaseID != "" {
|
|
qtRepo, err := postgres.NewQuestionTableExportRepository(store)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
qtDispatcher, err := buildQuestionTableDispatcher(qtRepo, dingCfg.ClientID, dingCfg.ClientSecret, qtCfg)
|
|
if err != nil {
|
|
log.Fatalf("DingTalk question table dispatcher init failed: %v", err)
|
|
}
|
|
questionTableDone = make(chan struct{})
|
|
go func() {
|
|
defer close(questionTableDone)
|
|
runQuestionTablePoller(ctx, qtDispatcher, 5*time.Second)
|
|
}()
|
|
log.Printf("DingTalk question table dispatcher started")
|
|
}
|
|
var streamClient *dingtalk.StreamClient
|
|
if dingCfg.Enabled() {
|
|
directoryLookup, err := directory.NewDingTalkLookup(directory.DingTalkLookupConfig{
|
|
TokenURL: "https://oapi.dingtalk.com/gettoken",
|
|
APIBaseURL: "https://oapi.dingtalk.com",
|
|
ClientID: dingCfg.ClientID,
|
|
ClientSecret: dingCfg.ClientSecret,
|
|
Timeout: 10 * time.Second,
|
|
})
|
|
if err != nil {
|
|
log.Fatalf("DingTalk directory lookup init failed: %v", err)
|
|
}
|
|
directoryResolver, err := directory.NewResolver(store, directoryLookup)
|
|
if err != nil {
|
|
log.Fatalf("DingTalk directory resolver init failed: %v", err)
|
|
}
|
|
outboundRewriter, err := buildOutboundImageRewriter(outboundImageConfig{
|
|
WeKnoraBaseURL: os.Getenv("WEKNORA_BASE_URL"),
|
|
WeKnoraAPIKey: os.Getenv("WEKNORA_API_KEY"),
|
|
Endpoint: os.Getenv("DOBOTAI_OUTBOUND_IMAGE_OSS_ENDPOINT"),
|
|
Region: os.Getenv("DOBOTAI_OUTBOUND_IMAGE_OSS_REGION"),
|
|
AccessKeyID: os.Getenv("DOBOTAI_OUTBOUND_IMAGE_OSS_ACCESS_KEY_ID"),
|
|
AccessKeySecret: os.Getenv("DOBOTAI_OUTBOUND_IMAGE_OSS_ACCESS_KEY_SECRET"),
|
|
Bucket: os.Getenv("DOBOTAI_OUTBOUND_IMAGE_OSS_BUCKET"),
|
|
})
|
|
if err != nil {
|
|
log.Fatalf("DingTalk outbound image init failed: %v", err)
|
|
}
|
|
feedbackService, err := feedback.NewService(dingCfg.TenantKey, "dingtalk", store)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
feedbackHandler, err := dingtalk.NewCardFeedbackHandler(feedbackService)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
sender, err := dingtalk.NewWebhookSender(nil)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
downloader, err := dingtalk.NewImageDownloader(dingCfg.ClientID, dingCfg.ClientSecret, nil)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
var cardSender dingtalk.FinalCardSender
|
|
if cardTemplateID := strings.TrimSpace(os.Getenv("DINGTALK_CARD_TEMPLATE_ID")); cardTemplateID != "" {
|
|
cardDebugUserID := os.Getenv("DINGTALK_CARD_DEBUG_USER_ID")
|
|
cs, err := dingtalk.NewCardSender(
|
|
dingCfg.ClientID,
|
|
dingCfg.ClientSecret,
|
|
cardTemplateID,
|
|
nil,
|
|
dingtalk.WithCardDebugUserID(cardDebugUserID),
|
|
)
|
|
if err != nil {
|
|
log.Fatalf("DingTalk card sender init failed: %v", err)
|
|
}
|
|
if strings.TrimSpace(cardDebugUserID) != "" {
|
|
log.Printf("DingTalk card debug enabled for one allowlisted user")
|
|
}
|
|
cardSender = cs
|
|
}
|
|
handler, err := dingtalk.NewHandlerWithOptions(
|
|
dingtalk.TenantConfig{TenantKey: dingCfg.TenantKey, Channel: "dingtalk"},
|
|
qaService,
|
|
sender,
|
|
dingtalk.HandlerOptions{
|
|
ImageDownloader: downloader,
|
|
CardSender: cardSender,
|
|
Directory: directoryResolver,
|
|
OutboundRewriter: outboundRewriter,
|
|
},
|
|
)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
streamClient, err = dingtalk.NewStreamClient(dingCfg, handler, feedbackHandler)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
if err := streamClient.Start(ctx); err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
defer streamClient.Close()
|
|
log.Printf("DingTalk Stream ingress started for tenant %s", dingCfg.TenantKey)
|
|
}
|
|
server := &http.Server{Addr: address, Handler: api.Handler(), ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 15 * time.Second, WriteTimeout: 130 * time.Second, IdleTimeout: 60 * time.Second}
|
|
go func() {
|
|
log.Printf("DobotAI ingress listening on %s", address)
|
|
if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
|
|
log.Printf("HTTP server failed: %v", err)
|
|
cancel()
|
|
}
|
|
}()
|
|
<-ctx.Done()
|
|
shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer shutdownCancel()
|
|
if err := server.Shutdown(shutdownCtx); err != nil {
|
|
log.Printf("HTTP shutdown: %v", err)
|
|
}
|
|
if faqWorkflowDone != nil {
|
|
select {
|
|
case <-faqWorkflowDone:
|
|
case <-shutdownCtx.Done():
|
|
log.Printf("FAQ table poller shutdown timed out")
|
|
}
|
|
}
|
|
if questionTableDone != nil {
|
|
select {
|
|
case <-questionTableDone:
|
|
case <-shutdownCtx.Done():
|
|
log.Printf("question table poller shutdown timed out")
|
|
}
|
|
}
|
|
}
|
|
|
|
func buildAuditOptions(repo audit.TurnRepository, tenantKey, adminToken, address string) (httpapi.Options, error) {
|
|
adminToken = strings.TrimSpace(adminToken)
|
|
if adminToken == "" {
|
|
return httpapi.Options{}, nil
|
|
}
|
|
if repo == nil {
|
|
return httpapi.Options{}, errors.New("audit repository is required when DOBOTAI_ADMIN_TOKEN is set")
|
|
}
|
|
tenantKey = strings.TrimSpace(tenantKey)
|
|
if tenantKey == "" {
|
|
return httpapi.Options{}, errors.New("DOBOTAI_TENANT_KEY is required when DOBOTAI_ADMIN_TOKEN is set")
|
|
}
|
|
if err := httpapi.ValidateLoopbackListenAddress(address); err != nil {
|
|
return httpapi.Options{}, err
|
|
}
|
|
service, err := audit.NewService(tenantKey, repo)
|
|
if err != nil {
|
|
return httpapi.Options{}, err
|
|
}
|
|
return httpapi.Options{AuditService: service, AdminToken: adminToken}, nil
|
|
}
|