hermes
6816289552
build / build (push) Successful in 1m23s
周报页新增自动发送区：管理员配置 SMTP/收件人/发送星期与小时，授权码用主密钥加密保存。 确认绑定当前草稿版本与正文哈希，正文再改则确认失效；worker 每分钟仅在启用、到点、周已结束且未发送过时发一次。 邮件同时提供渲染后的 HTML 与可读纯文本；SMTP 仅允许 TLS，拒绝明文。 迁移 000060 新增设置、确认与发送记录三张表，唯一键防重复发送。
251 lines
9.1 KiB
Go
251 lines
9.1 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log"
|
|
"os"
|
|
"os/signal"
|
|
"path/filepath"
|
|
"syscall"
|
|
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
|
|
"knowledge-governance/internal/appconfig"
|
|
"knowledge-governance/internal/governance"
|
|
"knowledge-governance/internal/jobs"
|
|
postgresstore "knowledge-governance/internal/platform/postgres"
|
|
"knowledge-governance/internal/platform/weknora"
|
|
"knowledge-governance/internal/versionhistory"
|
|
artifactfetch "knowledge-governance/internal/workflows/artifactfetch"
|
|
catalogrefresh "knowledge-governance/internal/workflows/catalogrefresh"
|
|
"knowledge-governance/internal/workflows/sourcescan"
|
|
"knowledge-governance/internal/workflows/testweknora"
|
|
)
|
|
|
|
func main() {
|
|
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
|
defer stop()
|
|
if err := run(ctx, os.Getenv); err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
}
|
|
|
|
func run(ctx context.Context, getenv func(string) string) error {
|
|
config, err := appconfig.Load(getenv)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
pool, err := pgxpool.New(ctx, config.DatabaseURL)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer pool.Close()
|
|
if err := pool.Ping(ctx); err != nil {
|
|
return err
|
|
}
|
|
if err := postgresstore.ApplyMigrations(ctx, pool); err != nil {
|
|
return err
|
|
}
|
|
store := postgresstore.NewStore(pool)
|
|
handlers, err := workerHandlers(store, config)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if syncErr := configureDirectorySync(pool, getenv, handlers); syncErr != nil {
|
|
log.Print("QA directory synchronization unavailable; handler was not registered")
|
|
}
|
|
mirrorStore, mirrorErr := configureRecordsSync(pool, getenv, config.DatabaseURL, handlers)
|
|
if mirrorErr != nil {
|
|
log.Print("QA local mirror configuration invalid; handler was not registered")
|
|
}
|
|
worker, err := jobs.NewWorker(jobs.WorkerConfig{
|
|
Queue: store,
|
|
WorkerID: config.WorkerID,
|
|
LeaseDuration: config.LeaseDuration,
|
|
RetryDelay: config.RetryDelay,
|
|
Handlers: handlers,
|
|
OnError: func(err error) { log.Printf("worker error: %v", err) },
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if ai := configureQAProcessing(pool, getenv, config.DatabaseURL); ai != nil {
|
|
go ai.Run(ctx, func(error) { log.Print("QA AI worker unavailable; durable work and original labels retained") })
|
|
}
|
|
if mirrorStore != nil {
|
|
go runRecordsSyncScheduler(ctx, mirrorStore, func(error) { log.Print("QA local mirror scheduler unavailable; last committed data retained") })
|
|
}
|
|
go weeklyMailLoop(ctx, pool, getenv)
|
|
go runSourceRefreshScheduler(ctx, store, config.SourceRefreshSchedulerInterval, func(err error) {
|
|
log.Printf("worker %s source refresh scheduler error: %v", config.WorkerID, err)
|
|
})
|
|
log.Printf("worker %s started (source refresh scheduler every %s)", config.WorkerID, config.SourceRefreshSchedulerInterval)
|
|
return worker.Run(ctx, config.PollInterval)
|
|
}
|
|
|
|
func workerHandlers(store *postgresstore.Store, config appconfig.Config) (map[string]jobs.Handler, error) {
|
|
handlers := map[string]jobs.Handler{
|
|
"demo": func(_ context.Context, job jobs.Job) error {
|
|
log.Printf("completed demo job id=%s", job.ID)
|
|
return nil
|
|
},
|
|
"demo.sleep": jobs.DemoSleepHandler,
|
|
}
|
|
catalogRunner, err := catalogrefresh.NewCommandRunner("/usr/bin/python3", "/app/scripts/live_catalog_refresh.py")
|
|
if err != nil {
|
|
return nil, fmt.Errorf("create catalog refresh runner: %w", err)
|
|
}
|
|
catalogHandler, err := catalogrefresh.NewHandler(catalogrefresh.Config{Store: store, Runner: catalogRunner})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("create catalog refresh handler: %w", err)
|
|
}
|
|
handlers["source.catalog.refresh"] = catalogHandler.Handle
|
|
ingestionRoot := config.IngestionSpoolRoot
|
|
if ingestionRoot == "" {
|
|
ingestionRoot = config.GovernanceSpoolRoot
|
|
}
|
|
if !filepath.IsAbs(ingestionRoot) || ingestionRoot == "" {
|
|
ingestionRoot = "/var/lib/governance/ingestion"
|
|
}
|
|
fetchMaxBytes := config.ArtifactFetchMaxBytes
|
|
if config.TestWeKnoraEnabled && config.TestWeKnoraMaxUploadBytes > 0 && fetchMaxBytes > config.TestWeKnoraMaxUploadBytes {
|
|
fetchMaxBytes = config.TestWeKnoraMaxUploadBytes
|
|
}
|
|
if fetchMaxBytes <= 0 {
|
|
fetchMaxBytes = 50 * 1024 * 1024
|
|
}
|
|
artifactRunner, err := artifactfetch.NewCommandRunner("/usr/bin/python3", "/app/scripts/artifact_fetch.py",
|
|
ingestionRoot, "ingestion", fetchMaxBytes)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("create artifact fetch runner: %w", err)
|
|
}
|
|
artifactHandler, err := artifactfetch.NewHandler(artifactfetch.Config{Store: store, Runner: artifactRunner})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("create artifact fetch handler: %w", err)
|
|
}
|
|
handlers["artifact.fetch"] = artifactHandler.Handle
|
|
if config.VersionHistoryGiteaBaseURL != "" {
|
|
// 「版本记录」同步：匿名只读远端仓库的提交/tag，写本库快照。未配置来源就不注册，
|
|
// 界面上的同步按钮也会显示"未接线"，不会入队一个没人处理的作业。
|
|
client, err := versionhistory.NewGiteaClient(versionhistory.GiteaConfig{
|
|
BaseURL: config.VersionHistoryGiteaBaseURL,
|
|
AllowedBaseURLs: config.VersionHistoryAllowedOrigins,
|
|
LinkOrigin: config.VersionHistoryLinkOrigin,
|
|
Repo: config.VersionHistoryRepo,
|
|
Timeout: config.VersionHistoryHTTPTimeout,
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("create version history client: %w", err)
|
|
}
|
|
syncer, err := versionhistory.NewSyncer(versionhistory.SyncerConfig{Source: client, Store: store,
|
|
RepoKey: config.VersionHistoryRepo, MaxPages: config.VersionHistoryMaxPages})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("create version history syncer: %w", err)
|
|
}
|
|
handlers[versionhistory.JobType] = func(ctx context.Context, job jobs.Job) error {
|
|
triggeredBy := ""
|
|
var payload struct {
|
|
TriggeredBy string `json:"triggered_by"`
|
|
TargetSHA string `json:"target_sha"`
|
|
}
|
|
if err := json.Unmarshal(job.Payload, &payload); err == nil {
|
|
triggeredBy = payload.TriggeredBy
|
|
}
|
|
activeSyncer := syncer
|
|
if payload.TargetSHA != "" {
|
|
var e error
|
|
activeSyncer, e = versionhistory.NewSyncer(versionhistory.SyncerConfig{Source: client, Store: store, RepoKey: config.VersionHistoryRepo, MaxPages: config.VersionHistoryMaxPages, TargetSHAs: []string{payload.TargetSHA}})
|
|
if e != nil {
|
|
return e
|
|
}
|
|
}
|
|
run, err := activeSyncer.Sync(ctx, job.ID, triggeredBy)
|
|
if err != nil {
|
|
log.Printf("version history sync failed job=%s run=%d code=%s: %v", job.ID, run.ID, run.ErrorCode, err)
|
|
return err
|
|
}
|
|
log.Printf("version history sync completed job=%s run=%d fetched=%d expected=%v complete=%v",
|
|
job.ID, run.ID, run.FetchedCount, run.ExpectedCount, run.Complete)
|
|
return nil
|
|
}
|
|
}
|
|
if config.PPScanEnabled {
|
|
scanner, err := sourcescan.NewFilePPScanner(sourcescan.FilePPScannerConfig{
|
|
AllowedRoot: config.PPScanAllowedRoot,
|
|
LedgerPath: config.PPScanLedgerPath,
|
|
ManifestPath: config.PPScanManifestPath,
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("create PP source scanner: %w", err)
|
|
}
|
|
handler, err := sourcescan.NewPPHandler(sourcescan.PPHandlerConfig{Store: store, Scanner: scanner})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("create PP source scan handler: %w", err)
|
|
}
|
|
handlers["source.scan.pp"] = handler.Handle
|
|
}
|
|
if !config.TestWeKnoraEnabled {
|
|
return handlers, nil
|
|
}
|
|
apiKey, err := weknora.ReadAPIKeyFile(config.TestWeKnoraAPIKeyFile)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("read test WeKnora API key: %w", err)
|
|
}
|
|
client, err := weknora.NewClient(weknora.ClientConfig{
|
|
BaseURL: config.TestWeKnoraBaseURL,
|
|
APIKey: apiKey,
|
|
Timeout: config.TestWeKnoraHTTPTimeout,
|
|
AllowDockerHTTP: config.TestWeKnoraAllowDockerHTTP,
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("create test WeKnora client: %w", err)
|
|
}
|
|
upload, err := testweknora.NewUploadHandler(testweknora.UploadHandlerConfig{
|
|
Store: store,
|
|
Client: client,
|
|
SpoolRoot: config.GovernanceSpoolRoot,
|
|
MaxUploadBytes: config.TestWeKnoraMaxUploadBytes,
|
|
FirstPollDelay: config.TestWeKnoraPollInitial,
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("create test upload handler: %w", err)
|
|
}
|
|
poll, err := testweknora.NewPollHandler(testweknora.PollHandlerConfig{
|
|
Store: store,
|
|
Client: client,
|
|
PollDelay: config.TestWeKnoraPollInitial,
|
|
StallAfter: config.TestWeKnoraStalledAfter,
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("create test poll handler: %w", err)
|
|
}
|
|
policy, err := governance.ParseAutoReviewPolicy(config.TestAutoReviewPolicy)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("resolve test auto review policy: %w", err)
|
|
}
|
|
validate, err := testweknora.NewValidateHandler(testweknora.ValidateHandlerConfig{
|
|
Store: store,
|
|
Client: client,
|
|
PolicyVersion: config.TestAcceptancePolicyVersion,
|
|
AutoReviewPolicy: policy,
|
|
Report: func(err error) { log.Printf("auto review error: %v", err) },
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("create test validate handler: %w", err)
|
|
}
|
|
remove, err := testweknora.NewRemoveHandler(testweknora.RemoveHandlerConfig{
|
|
Store: store,
|
|
Client: client,
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("create test removal handler: %w", err)
|
|
}
|
|
handlers["test.upload"] = upload.Handle
|
|
handlers["test.poll"] = poll.Handle
|
|
handlers["test.validate"] = validate.Handle
|
|
handlers["test.remove"] = remove.Handle
|
|
return handlers, nil
|
|
}
|