package bootstrap import ( "context" "crypto/tls" "errors" "fmt" "net/http" "os" "os/signal" "sync" "sync/atomic" "syscall" "time" "go.uber.org/zap" "spark-server/internal/infrastructure/id" "spark-server/internal/infrastructure/logger" "spark-server/internal/infrastructure/mysql" "spark-server/internal/infrastructure/objectstore" redisx "spark-server/internal/infrastructure/redis" adminlogic "spark-server/internal/module/admin/logic" adminmodel "spark-server/internal/module/admin/model" authlogic "spark-server/internal/module/auth/logic" convlogic "spark-server/internal/module/conversation/logic" convmodel "spark-server/internal/module/conversation/model" deliverlogic "spark-server/internal/module/deliver/logic" delivermodel "spark-server/internal/module/deliver/model" favlogic "spark-server/internal/module/favorite/logic" favmodel "spark-server/internal/module/favorite/model" friendlogic "spark-server/internal/module/friend/logic" friendmodel "spark-server/internal/module/friend/model" grouplogic "spark-server/internal/module/group/logic" groupmodel "spark-server/internal/module/group/model" inboxlogic "spark-server/internal/module/inbox/logic" inboxmodel "spark-server/internal/module/inbox/model" msglogic "spark-server/internal/module/message/logic" msgmodel "spark-server/internal/module/message/model" momentlogic "spark-server/internal/module/moment/logic" momentmodel "spark-server/internal/module/moment/model" smslogic "spark-server/internal/module/sms/logic" squarelogic "spark-server/internal/module/square/logic" squaremodel "spark-server/internal/module/square/model" synclogic "spark-server/internal/module/sync/logic" syncmodel "spark-server/internal/module/sync/model" uploadlogic "spark-server/internal/module/upload/logic" userlogic "spark-server/internal/module/user/logic" usermodel "spark-server/internal/module/user/model" "spark-server/internal/pkg/ratelimit" httpx "spark-server/internal/transport/http" tcp "spark-server/internal/transport/tcp" "spark-server/internal/transport/tcp/session" ) type App struct { cfg *Config log *zap.Logger logClose func() error http *httpx.Server tcp *tcp.Server pool tcp.TaskPool mysql *mysql.DB redis *redisx.Client idgen *id.Generator sessions session.Manager deliver *deliverlogic.Worker accepting atomic.Bool shutdownOnce sync.Once } func New(cfg *Config) (*App, error) { log, logClose := logger.New(&logger.Options{ Level: cfg.Log.Level, Encoding: cfg.Log.Encoding, Console: cfg.Log.Console, File: logger.FileOptions{ Path: cfg.Log.File.Path, MaxSizeMB: cfg.Log.File.MaxSizeMB, MaxBackups: cfg.Log.File.MaxBackups, MaxAgeDays: cfg.Log.File.MaxAgeDays, Compress: cfg.Log.File.Compress, }, }) idgen, err := id.New(cfg.ID.SnowflakeWorkerID) if err != nil { _ = logClose() return nil, err } db, err := mysql.Open(log, &mysql.Options{ DSN: cfg.MySQL.DSN, SlowThreshold: cfg.MySQL.SlowThreshold, LogLevel: cfg.MySQL.LogLevel, ZapLevel: cfg.Log.Level, MaxOpenConns: cfg.MySQL.MaxOpenConns, MaxIdleConns: cfg.MySQL.MaxIdleConns, ConnMaxLifetime: cfg.MySQL.ConnMaxLifetime, ConnMaxIdleTime: cfg.MySQL.ConnMaxIdleTime, }) if err != nil { _ = logClose() return nil, err } rdb, err := redisx.Open(&redisx.Options{ Addr: cfg.Redis.Addr, Username: cfg.Redis.Username, Password: cfg.Redis.Password, DB: cfg.Redis.DB, }) if err != nil { _ = db.Close() _ = logClose() return nil, err } return finishApp(cfg, log, logClose, idgen, db, rdb) } func finishApp(cfg *Config, log *zap.Logger, logClose func() error, idgen *id.Generator, db *mysql.DB, rdb *redisx.Client) (*App, error) { jwtSvc, err := authlogic.NewJWT(&authlogic.Options{ Secret: cfg.JWT.Secret, Issuer: cfg.JWT.Issuer, TTL: cfg.JWT.TTL, RefreshTTL: cfg.JWT.RefreshTTL, Denylist: jwtDenylist(rdb, cfg.JWT.DenylistPrefix), Pairs: jwtPairs(rdb, cfg.JWT.RefreshPrefix), }) if err != nil { _ = rdb.Close() _ = db.Close() _ = logClose() return nil, err } pool := tcp.NewTaskPoolRecover(cfg.TaskPool.Size, cfg.TaskPool.Queue, func(rec any) { log.Error("taskpool panic", zap.Any("panic", rec)) }) sessions := session.NewManager() tlsConf, err := loadTLS(cfg.Server.TCP.TLS) if err != nil { pool.Close() _ = rdb.Close() _ = db.Close() _ = logClose() return nil, err } authLimit := ratelimit.New(cfg.RateLimit.AuthPerSecond) objects, err := objectstore.Open(&objectstore.Options{ Endpoint: cfg.ObjectStorage.Endpoint, PublicEndpoint: cfg.ObjectStorage.PublicEndpoint, AccessKey: cfg.ObjectStorage.AccessKey, SecretKey: cfg.ObjectStorage.SecretKey, Bucket: cfg.ObjectStorage.Bucket, Region: cfg.ObjectStorage.Region, UseSSL: cfg.ObjectStorage.UseSSL, ChatExpireDays: cfg.ObjectStorage.ChatExpireDays, Log: log, }) if err != nil { pool.Close() _ = rdb.Close() _ = db.Close() _ = logClose() return nil, err } app := &App{cfg: cfg, log: log, logClose: logClose, pool: pool, mysql: db, redis: rdb, idgen: idgen, sessions: sessions} probe := &Probe{mysql: db, redis: rdb, objects: objects, pool: pool, accepting: &app.accepting, requireDeps: cfg.Production()} userRepo := wireUserRepo(db) smsSvc := wireSMS(cfg, rdb, userRepo) users := userlogic.NewService(userRepo, smsSvc, idgen, wireUserObjects(objects)) login := authlogic.NewLoginService(jwtSvc, users) login.Sessions = sessions var deliverQueue deliverlogic.Queue inbox, mailbox := wireInbox(idgen, db, rdb) app.deliver, deliverQueue = wireDeliver(cfg, log, rdb, sessions, inbox, mailbox) syncSvc := synclogic.NewService(inbox, mailbox) chat := wireChat(idgen, db) friends := wireFriends(idgen, db, users, deliverQueue, chat) groups := wireGroups(idgen, db, users, friends, deliverQueue, objects, chat) messages := wireMessages(idgen, friends, groups, deliverQueue, chat) files := wireFiles(cfg, objects, groups) favorites := wireFavorites(idgen, db, objects, chat) moments := wireMoments(idgen, db, objects, friends, users) square := wireSquare(idgen, db, objects, friends, users) users.BindAccountCleanup(jwtSvc, sessions, accountCleaners(friends, moments, square, groups, favorites, inbox, mailbox, chat)...) adminSvc, adminTokens := wireAdmin(cfg, db, groups, idgen) app.tcp = tcp.NewServer(log, pool, sessions, &tcp.Options{ Addr: cfg.Server.TCP.Addr, Multicore: cfg.Server.TCP.Multicore, KeepAlive: cfg.Server.TCP.KeepAlive, RecvBuffer: cfg.Server.TCP.RecvBuffer, SendBuffer: cfg.Server.TCP.SendBuffer, MaxFrameSize: cfg.Server.TCP.MaxFrameSize, TLS: tlsConf, ProxyProtocol: cfg.Server.TCP.ProxyProtocol, IdleTimeout: cfg.Server.TCP.IdleTimeout, OutboundSize: cfg.Session.OutboundQueueSize, RatePerSec: cfg.RateLimit.MessagePerSecond, CompressMin: cfg.Server.TCP.CompressMinSize, IDs: idgen, Auth: jwtSvc, AuthLimiter: authLimit, Messages: messages, Sync: syncSvc, }) app.http = httpx.New(log, cfg.Server.HTTP.Addr, httpEngineDeps( jwtSvc, login, smsSvc, users, friends, groups, files, favorites, moments, square, messages, syncSvc, adminSvc, adminTokens, probe, authLimit, )) return app, nil } func httpEngineDeps( jwtSvc *authlogic.JWT, login *authlogic.LoginService, sms *smslogic.Service, users *userlogic.Service, friends *friendlogic.Service, groups *grouplogic.Service, files *uploadlogic.Service, favorites *favlogic.Service, moments *momentlogic.Service, square *squarelogic.Service, messages *msglogic.Service, syncSvc *synclogic.Service, adminSvc *adminlogic.Service, adminTokens *adminlogic.TokenService, probe *Probe, authLimit *ratelimit.Limiter, ) *httpx.EngineDeps { return &httpx.EngineDeps{ Auth: jwtSvc, Login: login, SMS: sms, Users: users, Friends: friends, Groups: groups, Files: files, Favorites: favorites, Moments: moments, Square: square, Messages: messages, Sync: syncSvc, Admin: adminSvc, AdminTokens: adminTokens, Health: probe, AuthLimiter: authLimit, } } func wireAdmin(cfg *Config, db *mysql.DB, groups *grouplogic.Service, idgen *id.Generator) (*adminlogic.Service, *adminlogic.TokenService) { tokens := adminlogic.NewTokenService(cfg.Admin.JWTSecret) var repo *adminmodel.Repo if db != nil && db.Ready() { repo = adminmodel.NewRepo(db.Gorm()) } return adminlogic.NewService(repo, groups, idgen), tokens } func loadTLS(t TLSConfig) (*tls.Config, error) { if t.CertFile == "" || t.KeyFile == "" { return nil, nil } cert, err := tls.LoadX509KeyPair(t.CertFile, t.KeyFile) if err != nil { return nil, fmt.Errorf("tls: %w", err) } return &tls.Config{MinVersion: tls.VersionTLS12, Certificates: []tls.Certificate{cert}}, nil } func wireUserRepo(db *mysql.DB) userlogic.Repository { if db != nil && db.Ready() { return usermodel.NewRepo(db.Gorm()) } return usermodel.NewMemoryRepo() } func wireUserObjects(store *objectstore.Store) userlogic.Objects { if store != nil && store.Ready() { return store } return nil } type chatStores struct { msgs msglogic.Repository chats msglogic.Conversations p2p *convlogic.P2PService } func wireChat(idgen *id.Generator, db *mysql.DB) chatStores { if idgen == nil { return chatStores{} } if db != nil && db.Ready() { convRepo := convmodel.NewRepo(db.Gorm()) return chatStores{msgs: msgmodel.NewRepo(db.Gorm()), chats: convRepo, p2p: convlogic.NewP2PService(convRepo, idgen)} } mem := convmodel.NewMemoryP2P() return chatStores{msgs: msgmodel.NewMemoryRepo(), chats: mem, p2p: convlogic.NewP2PService(mem, idgen)} } func wireFriends(idgen *id.Generator, db *mysql.DB, users *userlogic.Service, q friendlogic.DeliverQueue, chat chatStores) *friendlogic.Service { var repo friendlogic.Repository if db != nil && db.Ready() { repo = friendmodel.NewRepo(db.Gorm()) } else { repo = friendmodel.NewMemoryRepo() } return friendlogic.NewService(repo, users, idgen, friendlogic.NewGreeter(chat.p2p, chat.msgs, idgen, q)) } func wireGroups(idgen *id.Generator, db *mysql.DB, users *userlogic.Service, friends *friendlogic.Service, q grouplogic.DeliverQueue, store *objectstore.Store, chat chatStores) *grouplogic.Service { var repo grouplogic.Repository if db != nil && db.Ready() { repo = groupmodel.NewRepo(db.Gorm()) } else { repo = groupmodel.NewMemoryRepo() } var objects grouplogic.Objects if store != nil && store.Ready() { objects = store } return grouplogic.NewService(repo, users, friends, idgen, chat.msgs, q, objects) } func wireFiles(cfg *Config, store *objectstore.Store, groups *grouplogic.Service) *uploadlogic.Service { if store == nil || !store.Ready() { return nil } return uploadlogic.NewService(store, uploadlogic.Options{ TTL: cfg.ObjectStorage.PresignTTL, Limit: ratelimit.New(cfg.RateLimit.UploadPerSecond), Groups: groups, }) } func wireFavorites(idgen *id.Generator, db *mysql.DB, store *objectstore.Store, chat chatStores) *favlogic.Service { if idgen == nil { return nil } var repo favlogic.Repository if db != nil && db.Ready() { repo = favmodel.NewRepo(db.Gorm()) } else { repo = favmodel.NewMemoryRepo() } var objects favlogic.Objects if store != nil && store.Ready() { objects = store } return favlogic.NewService(repo, chat.msgs, chat.chats, objects, idgen) } func wireSquare(idgen *id.Generator, db *mysql.DB, store *objectstore.Store, friends *friendlogic.Service, users *userlogic.Service) *squarelogic.Service { if idgen == nil || friends == nil || users == nil { return nil } var repo squarelogic.Repository if db != nil && db.Ready() { repo = squaremodel.NewRepo(db.Gorm()) } else { repo = squaremodel.NewMemoryRepo() } var objects squarelogic.Objects if store != nil && store.Ready() { objects = store } return squarelogic.NewService(repo, friends, users, objects, idgen) } func wireMoments(idgen *id.Generator, db *mysql.DB, store *objectstore.Store, friends *friendlogic.Service, users *userlogic.Service) *momentlogic.Service { if idgen == nil || friends == nil || users == nil { return nil } var repo momentlogic.Repository if db != nil && db.Ready() { repo = momentmodel.NewRepo(db.Gorm()) } else { repo = momentmodel.NewMemoryRepo() } var objects momentlogic.Objects if store != nil && store.Ready() { objects = store } return momentlogic.NewService(repo, friends, users, objects, idgen) } func wireInbox(idgen *id.Generator, db *mysql.DB, rdb *redisx.Client) (inboxlogic.Store, synclogic.Mailbox) { var inbox inboxlogic.Store if rdb != nil && rdb.Ready() { inbox = inboxmodel.NewRedisStore(rdb.RDB()) } else { inbox = inboxmodel.NewMemoryStore() } var box synclogic.Mailbox if db != nil && db.Ready() { box = syncmodel.NewStore(db.Gorm(), idgen) } else { box = syncmodel.NewMemoryStore(idgen) } return inbox, box } func wireDeliver(cfg *Config, log *zap.Logger, rdb *redisx.Client, sessions session.Manager, inbox inboxlogic.Store, mailbox deliverlogic.Mailbox) (*deliverlogic.Worker, deliverlogic.Queue) { var q deliverlogic.Queue var done deliverlogic.Deduper if rdb != nil && rdb.Ready() { q = delivermodel.NewRedisQueue(rdb.RDB(), cfg.Redis.DeliverQueueKey, cfg.Redis.DeliverProcessingKey) done = delivermodel.NewRedisDone(rdb.RDB(), cfg.Redis.DeliverDonePrefix, cfg.Deliver.DoneTTL) } else { q = delivermodel.NewMemoryQueue() done = delivermodel.NewMemoryDone() } return deliverlogic.NewWorker(log, q, done, deliverlogic.WorkerOptions{ Workers: cfg.Deliver.Workers, TakeTimeout: cfg.Deliver.TakeTimeout, Handler: deliverlogic.InboxFanout(sessions, inbox, mailbox), }), q } func wireMessages(idgen *id.Generator, friends *friendlogic.Service, groups *grouplogic.Service, q msglogic.DeliverQueue, chat chatStores) *msglogic.Service { if idgen == nil { return nil } return msglogic.NewService(chat.msgs, chat.chats, chat.p2p, friends, groups, idgen, q) } func accountCleaners( friends *friendlogic.Service, moments *momentlogic.Service, square *squarelogic.Service, groups *grouplogic.Service, favorites *favlogic.Service, inbox inboxlogic.Store, mailbox synclogic.Mailbox, chat chatStores, ) []userlogic.UserDataCleaner { out := make([]userlogic.UserDataCleaner, 0, 8) if friends != nil { out = append(out, friends) } if moments != nil { out = append(out, moments) } if square != nil { out = append(out, square) } if groups != nil { out = append(out, groups) } if favorites != nil { out = append(out, favorites) } if inbox != nil { out = append(out, userlogic.CleanerFunc(inbox.PurgeUser)) } if mailbox != nil { out = append(out, userlogic.CleanerFunc(mailbox.PurgeUser)) } if chat.msgs != nil { out = append(out, userlogic.CleanerFunc(chat.msgs.PurgeUser)) } if p, ok := chat.chats.(interface { PurgeUser(context.Context, int64) error }); ok { out = append(out, userlogic.CleanerFunc(p.PurgeUser)) } return out } func jwtDenylist(rdb *redisx.Client, prefix string) authlogic.Denylist { if rdb != nil && rdb.Ready() { return authlogic.NewRedisDenylist(rdb.RDB(), prefix) } return authlogic.NewMemoryDenylist() } func jwtPairs(rdb *redisx.Client, prefix string) authlogic.PairStore { if rdb != nil && rdb.Ready() { return authlogic.NewRedisPairs(rdb.RDB(), prefix) } return authlogic.NewMemoryPairs() } func (a *App) Run() error { if a.deliver != nil { a.deliver.Start(context.Background()) } errCh := make(chan error, 2) go func() { errCh <- a.http.Start() }() go func() { errCh <- a.tcp.Start() }() if err := a.waitListening(errCh); err != nil { return errors.Join(err, a.Shutdown()) } a.accepting.Store(true) a.log.Info("ready") return a.serve(errCh) } func (a *App) waitListening(errCh <-chan error) error { ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) defer cancel() done := make(chan error, 1) go func() { done <- errors.Join(a.http.WaitReady(ctx), a.tcp.WaitReady(ctx)) }() select { case err := <-errCh: if err == nil { return ctx.Err() } return err case err := <-done: return err } } func (a *App) serve(errCh <-chan error) error { sigCh := make(chan os.Signal, 1) signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM) defer signal.Stop(sigCh) select { case err := <-errCh: a.log.Error("server exit", zap.Error(err)) return errors.Join(err, a.Shutdown()) case sig := <-sigCh: a.log.Info("shutdown", zap.String("signal", sig.String())) return a.Shutdown() } } func (a *App) Shutdown() error { var err error a.shutdownOnce.Do(func() { err = a.doShutdown() }) return err } func (a *App) doShutdown() error { a.accepting.Store(false) ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) defer cancel() var err error if e := a.http.Stop(ctx); e != nil && !errors.Is(e, http.ErrServerClosed) { err = errors.Join(err, e) } if e := a.tcp.Stop(ctx); e != nil { err = errors.Join(err, e) } if a.pool != nil { a.pool.Close() } if a.deliver != nil { a.deliver.Stop() } err = errors.Join(err, a.redis.Close(), a.mysql.Close()) if a.logClose != nil { err = errors.Join(err, a.logClose()) } return err }