feat(deploy): 在线更新 - 新增后台检查、强制备份迁移与失败自动回退

2026-08-30 22:15:17 +08:00
parent 4ba9694f45
commit db66ac1182
60 changed files with 4124 additions and 93 deletions
--- a/backend/cmd/server/main.go
+++ b/backend/cmd/server/main.go
@@ -1,26 +1,41 @@
 package main

 import (
+	"context"
+	"errors"
 	"fmt"
 	"log"
+	"net"
 	"net/http"
 	"net/url"
 	"os"
+	"os/signal"
+	"strconv"
 	"strings"
+	"syscall"
 	"time"

 	"infinite-canvas/backend/internal/database"
 	"infinite-canvas/backend/internal/handler"
 	"infinite-canvas/backend/internal/repository"
 	"infinite-canvas/backend/internal/service"
+	"infinite-canvas/backend/internal/updaterclient"

 	"github.com/gin-gonic/gin"
 )

 func main() {
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer stop()
+	if err := run(ctx); err != nil {
+		log.Fatal(err)
+	}
+}
+
+func run(ctx context.Context) error {
 	dataDir := env("CANVAS_BACKEND_DATA_DIR", "data")
 	if err := os.MkdirAll(dataDir, 0o755); err != nil {
-		log.Fatal(err)
+		return err
 	}
 	db, err := database.Open(database.Config{
 		Driver:  env("CANVAS_DATABASE_DRIVER", "sqlite"),
@@ -28,44 +43,60 @@ func main() {
 		DataDir: dataDir,
 	})
 	if err != nil {
-		log.Fatal(err)
+		return err
 	}
 	if err := database.ConfigurePool(db); err != nil {
-		log.Fatal(err)
+		return err
 	}
-	if err := database.MigrateSchema(db); err != nil {
-		log.Fatal(err)
+	sqlDB, err := db.DB()
+	if err != nil {
+		return err
+	}
+	defer sqlDB.Close()
+	autoMigrate, err := envBool("CANVAS_AUTO_MIGRATE", true)
+	if err != nil {
+		return err
+	}
+	if autoMigrate {
+		err = database.MigrateSchema(db)
+	} else {
+		err = database.RequireSchemaVersion(db)
+	}
+	if err != nil {
+		return err
 	}

 	repo := repository.New(db)
 	addr := env("CANVAS_BACKEND_ADDR", ":8080")
 	capabilities := service.RuntimeCapabilitiesForDeployment(addr, os.Getenv("CANVAS_DESKTOP_LOCAL_CHANNELS_ENABLED"))
 	svc := service.NewWithRuntimeCapabilities(repo, dataDir, capabilities)
+	if updaterToken := strings.TrimSpace(os.Getenv("CANVAS_UPDATER_TOKEN")); updaterToken != "" {
+		svc.ConfigureUpdateManager(updaterclient.New(env("CANVAS_UPDATER_SOCKET", "/run/open-ai-canvas-updater/updater.sock"), updaterToken))
+	}
+	defer svc.Close()
 	if err := svc.ValidateRuntime(); err != nil {
-		log.Fatal(err)
+		return err
 	}
 	if err := svc.EnsureSystemChannelModels(); err != nil {
-		log.Fatal(err)
+		return err
 	}
 	if err := svc.EnsureDefaultPromptTemplates(); err != nil {
-		log.Fatal(err)
+		return err
 	}
 	if err := svc.EnsureBuiltinProjectWorkflowTemplate(); err != nil {
-		log.Fatal(err)
+		return err
 	}
 	if err := svc.EnsureBuiltinSkills(); err != nil {
-		log.Fatal(err)
+		return err
 	}
 	if err := svc.EnsureSkillPackages(); err != nil {
-		log.Fatal(err)
+		return err
 	}
 	if summary, err := svc.MigrateLegacyStorage(); err != nil {
 		log.Printf("storage migration skipped after error: %v", err)
 	} else if summary.Backup != "" {
 		log.Printf("storage migration completed: tasks=%d assets=%d projects=%d backup=%s", summary.Tasks, summary.Assets, summary.Projects, summary.Backup)
 	}
-	svc.StartWorker()
-
 	r := gin.New()
 	r.Use(gin.LoggerWithFormatter(func(param gin.LogFormatterParams) string {
 		return fmt.Sprintf("%s - [%s] \"%s %s\" %d %s %s\n", param.ClientIP, param.TimeStamp.Format(time.RFC3339), param.Method, redactCanvasSharePath(param.Path), param.StatusCode, param.Latency, param.ErrorMessage)
@@ -73,14 +104,13 @@ func main() {
 	r.Use(handler.RequestCorrelationMiddleware())
 	corsMiddleware, err := cors()
 	if err != nil {
-		log.Fatal(err)
+		return err
 	}
 	r.Use(corsMiddleware)
 	handler.ConfigureRuntime(svc)
 	api := r.Group("/api")
-	api.GET("/health", func(c *gin.Context) {
-		c.JSON(200, gin.H{"code": 0, "data": gin.H{"status": "ok"}, "msg": "ok"})
-	})
+	status := newSystemStatus(db, svc)
+	registerSystemStatusRoutes(api, status)
 	handler.RegisterOAuthCallbackRoutes(r, svc)
 	handler.RegisterAuthRoutes(api, svc)
 	handler.RegisterFeatureAvailabilityRoutes(api, svc)
@@ -88,6 +118,7 @@ func main() {
 	handler.RegisterAdminRoutes(api, svc)
 	handler.RegisterAdminAnalyticsRoutes(api, svc)
 	handler.RegisterAdminStorageRoutes(api, svc)
+	handler.RegisterAdminUpdateRoutes(api, svc)
 	handler.RegisterAnnouncementRoutes(api, svc)
 	handler.RegisterFinanceRoutes(api, svc)
 	handler.RegisterLibTVRoutes(api, svc)
@@ -112,10 +143,52 @@ func main() {
 	handler.RegisterCanvasShareRoutes(api, svc)
 	r.NoRoute(handler.SystemProxyNoRouteHandler(svc))

-	log.Printf("影策 backend listening on %s", addr)
-	if err := r.Run(addr); err != nil {
-		log.Fatal(err)
+	listener, err := net.Listen("tcp", addr)
+	if err != nil {
+		return err
 	}
+	workerTimeout, err := envDuration("CANVAS_SHUTDOWN_TIMEOUT", 10*time.Minute)
+	if err != nil {
+		_ = listener.Close()
+		return err
+	}
+	httpServer := &http.Server{Handler: r, ReadHeaderTimeout: 10 * time.Second}
+	svc.StartWorker()
+	status.markStarted()
+	serveErr := make(chan error, 1)
+	go func() { serveErr <- httpServer.Serve(listener) }()
+	log.Printf("影策 backend listening on %s", addr)
+
+	var serveFailure error
+	select {
+	case <-ctx.Done():
+	case err := <-serveErr:
+		if err != nil && !errors.Is(err, http.ErrServerClosed) {
+			serveFailure = fmt.Errorf("HTTP 服务异常退出：%w", err)
+		}
+	}
+
+	status.beginDrain()
+	var shutdownFailures []error
+	if serveFailure != nil {
+		shutdownFailures = append(shutdownFailures, serveFailure)
+	}
+	httpShutdownCtx, httpShutdownCancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer httpShutdownCancel()
+	if err := httpServer.Shutdown(httpShutdownCtx); err != nil {
+		_ = httpServer.Close()
+		shutdownFailures = append(shutdownFailures, fmt.Errorf("关闭 HTTP 服务：%w", err))
+	}
+	workerCtx, workerCancel := context.WithTimeout(context.Background(), workerTimeout)
+	defer workerCancel()
+	if err := svc.StopWorker(workerCtx); err != nil {
+		shutdownFailures = append(shutdownFailures, fmt.Errorf("等待后台任务退出：%w", err))
+	}
+	if err := errors.Join(shutdownFailures...); err != nil {
+		return err
+	}
+	log.Printf("影策 backend stopped gracefully")
+	return nil
 }

 func redactCanvasSharePath(path string) string {
@@ -138,6 +211,30 @@ func env(key string, fallback string) string {
 	return value
 }

+func envBool(key string, fallback bool) (bool, error) {
+	value := strings.TrimSpace(os.Getenv(key))
+	if value == "" {
+		return fallback, nil
+	}
+	parsed, err := strconv.ParseBool(value)
+	if err != nil {
+		return false, fmt.Errorf("%s 必须是 true 或 false", key)
+	}
+	return parsed, nil
+}
+
+func envDuration(key string, fallback time.Duration) (time.Duration, error) {
+	value := strings.TrimSpace(os.Getenv(key))
+	if value == "" {
+		return fallback, nil
+	}
+	parsed, err := time.ParseDuration(value)
+	if err != nil || parsed <= 0 {
+		return 0, fmt.Errorf("%s 必须是正数时长，例如 10m", key)
+	}
+	return parsed, nil
+}
+
 const corsAllowedHeaders = "Accept, Content-Type, Authorization, X-Requested-With, X-Canvas-Scene, X-Idempotency-Key, X-Canvas-Trace-ID, X-Canvas-Upstream-URL, X-Canvas-Upstream-Format, X-Canvas-Allow-Local-Channel, X-Canvas-Upstream-Base-URL"

 const corsAllowedMethods = "GET, POST, PUT, PATCH, DELETE, OPTIONS"
--- a/backend/cmd/server/main_test.go
+++ b/backend/cmd/server/main_test.go
@@ -3,6 +3,7 @@ package main
 import (
 	"net/http/httptest"
 	"testing"
+	"time"

 	"github.com/gin-gonic/gin"
 )
@@ -19,6 +20,19 @@ func TestAllowedOriginWildcard(t *testing.T) {
 	}
 }

+func TestEnvironmentParsers(t *testing.T) {
+	t.Setenv("CANVAS_AUTO_MIGRATE", "false")
+	value, err := envBool("CANVAS_AUTO_MIGRATE", true)
+	if err != nil || value {
+		t.Fatalf("envBool = %v, %v", value, err)
+	}
+	t.Setenv("CANVAS_SHUTDOWN_TIMEOUT", "45s")
+	duration, err := envDuration("CANVAS_SHUTDOWN_TIMEOUT", time.Minute)
+	if err != nil || duration != 45*time.Second {
+		t.Fatalf("envDuration = %v, %v", duration, err)
+	}
+}
+
 func TestAllowedOriginUsesForwardedHost(t *testing.T) {
 	t.Setenv("CANVAS_CORS_ORIGINS", "")
 	context, _ := gin.CreateTestContext(httptest.NewRecorder())
--- a/backend/cmd/server/system_status.go
+++ b/backend/cmd/server/system_status.go
@@ -0,0 +1,112 @@
+package main
+
+import (
+	"context"
+	"net/http"
+	"sync/atomic"
+
+	"infinite-canvas/backend/internal/buildinfo"
+	"infinite-canvas/backend/internal/database"
+	"infinite-canvas/backend/internal/service"
+
+	"github.com/gin-gonic/gin"
+	"gorm.io/gorm"
+)
+
+type systemStatus struct {
+	db       *gorm.DB
+	service  *service.Service
+	started  atomic.Bool
+	draining atomic.Bool
+}
+
+type systemStatusSnapshot struct {
+	Status            string                `json:"status"`
+	Ready             bool                  `json:"ready"`
+	Started           bool                  `json:"started"`
+	Draining          bool                  `json:"draining"`
+	ActiveWorkerTasks int64                 `json:"activeWorkerTasks"`
+	Build             buildinfo.Info        `json:"build"`
+	Schema            database.SchemaStatus `json:"schema"`
+	Checks            systemStatusChecks    `json:"checks"`
+}
+
+type systemStatusChecks struct {
+	Database bool `json:"database"`
+	Runtime  bool `json:"runtime"`
+	Schema   bool `json:"schema"`
+}
+
+func newSystemStatus(db *gorm.DB, svc *service.Service) *systemStatus {
+	return &systemStatus{db: db, service: svc}
+}
+
+func (s *systemStatus) markStarted() { s.started.Store(true) }
+
+func (s *systemStatus) beginDrain() {
+	s.draining.Store(true)
+	if s.service != nil {
+		s.service.BeginDrain()
+	}
+}
+
+func (s *systemStatus) snapshot(ctx context.Context) systemStatusSnapshot {
+	snapshot := systemStatusSnapshot{
+		Started:  s.started.Load(),
+		Draining: s.draining.Load(),
+		Build:    buildinfo.Current(),
+		Schema:   database.SchemaStatus{Expected: database.CurrentSchemaVersion},
+	}
+	if s.service != nil {
+		snapshot.ActiveWorkerTasks = s.service.ActiveWorkerTasks()
+		snapshot.Checks.Runtime = s.service.ValidateRuntime() == nil
+	}
+	if s.db != nil {
+		snapshot.Checks.Database = s.db.WithContext(ctx).Exec("SELECT 1").Error == nil
+		if schema, err := database.ReadSchemaStatus(s.db); err == nil {
+			snapshot.Schema = schema
+			snapshot.Checks.Schema = schema.Ready
+		}
+	}
+	snapshot.Ready = snapshot.Started && !snapshot.Draining && snapshot.Checks.Database && snapshot.Checks.Runtime && snapshot.Checks.Schema
+	switch {
+	case snapshot.Draining:
+		snapshot.Status = "draining"
+	case snapshot.Ready:
+		snapshot.Status = "ok"
+	case !snapshot.Started:
+		snapshot.Status = "starting"
+	default:
+		snapshot.Status = "unhealthy"
+	}
+	return snapshot
+}
+
+func registerSystemStatusRoutes(api *gin.RouterGroup, status *systemStatus) {
+	api.GET("/health/live", func(c *gin.Context) {
+		c.JSON(http.StatusOK, gin.H{"code": 0, "data": gin.H{"status": "ok", "build": buildinfo.Current()}, "msg": "ok"})
+	})
+	startup := func(c *gin.Context) {
+		snapshot := status.snapshot(c.Request.Context())
+		if !snapshot.Started {
+			c.JSON(http.StatusServiceUnavailable, gin.H{"code": http.StatusServiceUnavailable, "data": snapshot, "msg": "服务仍在启动"})
+			return
+		}
+		c.JSON(http.StatusOK, gin.H{"code": 0, "data": snapshot, "msg": "ok"})
+	}
+	ready := func(c *gin.Context) {
+		snapshot := status.snapshot(c.Request.Context())
+		if !snapshot.Ready {
+			c.JSON(http.StatusServiceUnavailable, gin.H{"code": http.StatusServiceUnavailable, "data": snapshot, "msg": "服务暂未就绪"})
+			return
+		}
+		c.JSON(http.StatusOK, gin.H{"code": 0, "data": snapshot, "msg": "ok"})
+	}
+	api.GET("/health/startup", startup)
+	api.GET("/health/ready", ready)
+	api.GET("/health", ready)
+	api.GET("/system/version", func(c *gin.Context) {
+		snapshot := status.snapshot(c.Request.Context())
+		c.JSON(http.StatusOK, gin.H{"code": 0, "data": gin.H{"build": snapshot.Build, "schema": snapshot.Schema}, "msg": "ok"})
+	})
+}
--- a/backend/cmd/server/system_status_test.go
+++ b/backend/cmd/server/system_status_test.go
@@ -0,0 +1,60 @@
+package main
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"infinite-canvas/backend/internal/database"
+	"infinite-canvas/backend/internal/repository"
+	"infinite-canvas/backend/internal/service"
+
+	"github.com/gin-gonic/gin"
+)
+
+func TestSystemReadinessTracksStartupAndDrain(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	db, err := database.Open(database.Config{Driver: "sqlite", DSN: "file:system-status?mode=memory&cache=shared"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := database.MigrateSchema(db); err != nil {
+		t.Fatal(err)
+	}
+	svc := service.New(repository.New(db), t.TempDir())
+	status := newSystemStatus(db, svc)
+	router := gin.New()
+	registerSystemStatusRoutes(router.Group("/api"), status)
+
+	assertStatus(t, router, "/api/health/live", http.StatusOK)
+	assertStatus(t, router, "/api/health/ready", http.StatusServiceUnavailable)
+	status.markStarted()
+	assertStatus(t, router, "/api/health/ready", http.StatusOK)
+	status.beginDrain()
+	assertStatus(t, router, "/api/health/ready", http.StatusServiceUnavailable)
+	assertStatus(t, router, "/api/health/live", http.StatusOK)
+}
+
+func TestSystemReadinessRejectsMissingSchemaVersion(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	db, err := database.Open(database.Config{Driver: "sqlite", DSN: "file:system-status-schema?mode=memory&cache=shared"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	svc := service.New(repository.New(db), t.TempDir())
+	status := newSystemStatus(db, svc)
+	status.markStarted()
+	router := gin.New()
+	registerSystemStatusRoutes(router.Group("/api"), status)
+	assertStatus(t, router, "/api/health/ready", http.StatusServiceUnavailable)
+}
+
+func assertStatus(t *testing.T, handler http.Handler, path string, want int) {
+	t.Helper()
+	request := httptest.NewRequest(http.MethodGet, path, nil)
+	recorder := httptest.NewRecorder()
+	handler.ServeHTTP(recorder, request)
+	if recorder.Code != want {
+		t.Fatalf("GET %s status = %d, want %d; body=%s", path, recorder.Code, want, recorder.Body.String())
+	}
+}