spark-server/internal/bootstrap/app.go
DESKTOP-L8OUL9F\gfl0034 5fda6be8ec feat(squarebot): 接入媒体引导与 Bot 附件/锁
完善 bootstrap 媒体装配、配置开关,并补充 squarebot 附件与锁逻辑。
2026-09-24 23:37:44 +08:00

638 lines
20 KiB
Go

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"
clientlogic "spark-server/internal/module/client/logic"
clientmodel "spark-server/internal/module/client/model"
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"
devicelogic "spark-server/internal/module/device/logic"
devicemodel "spark-server/internal/module/device/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"
musiclogic "spark-server/internal/module/music/logic"
musicmodel "spark-server/internal/module/music/model"
smslogic "spark-server/internal/module/sms/logic"
squarelogic "spark-server/internal/module/square/logic"
squaremodel "spark-server/internal/module/square/model"
squarebotlogic "spark-server/internal/module/squarebot/logic"
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
squareBot *squarebotlogic.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, authlogic.PairKeyMode(cfg.JWT.PairKeyMode)),
PairKeyMode: authlogic.PairKeyMode(cfg.JWT.PairKeyMode),
})
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))
devices := wireDevices(db)
login := authlogic.NewLoginService(jwtSvc, users)
login.Devices = devices
login.Sessions = sessions
client := wireClient(db, users)
if client != nil {
login.Passwords = client
}
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, log, objects, groups)
favorites := wireFavorites(idgen, db, objects, chat)
moments := wireMoments(idgen, db, objects, friends, users)
square := wireSquare(idgen, db, objects, friends, users)
mediaPol := MediaPolicy(cfg.Media)
if moments != nil {
moments.ConfigureMedia(mediaPol, log)
}
if square != nil {
square.ConfigureMedia(mediaPol, log)
}
app.squareBot = wireSquareBot(cfg, log, rdb, objects, square, users)
music := wireMusic(cfg, db, objects)
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, music, client, 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,
music *musiclogic.Service,
client *clientlogic.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, Music: music, Client: client,
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, log *zap.Logger, 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,
Media: MediaPolicy(cfg.Media),
Log: log,
})
}
func wireSquareBot(
cfg *Config, log *zap.Logger, rdb *redisx.Client, store *objectstore.Store,
square *squarelogic.Service, users *userlogic.Service,
) *squarebotlogic.Worker {
if cfg == nil || !cfg.SquareBot.Enabled || square == nil || users == nil {
return nil
}
var locker squarebotlogic.Locker
if rdb != nil && rdb.Ready() {
locker = squarebotlogic.NewRedisLocker(rdb.RDB())
}
svc := squarebotlogic.NewService(log, square, users, locker, squarebotlogic.Options{
Enabled: true, Interval: cfg.SquareBot.Interval, TTL: cfg.SquareBot.TTL,
Feeds: cfg.SquareBot.Feeds, BotIndexes: cfg.SquareBot.BotIndexes,
})
if cfg.Media.BotCompress && store != nil && store.Ready() {
lim, jq, crf, ff := MediaCompressOptions(cfg.Media)
svc.BindMedia(squarebotlogic.NewMediaCompressor(lim, jq, crf, ff), store, true)
}
return squarebotlogic.NewWorker(log, svc)
}
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 wireMusic(cfg *Config, db *mysql.DB, store *objectstore.Store) *musiclogic.Service {
if db == nil || !db.Ready() {
return nil
}
var objects musiclogic.Objects
if store != nil && store.Ready() {
objects = store
}
publicBase := ""
bucket := "spark"
if cfg != nil {
publicBase = cfg.ObjectStorage.PublicEndpoint
if cfg.ObjectStorage.Bucket != "" {
bucket = cfg.ObjectStorage.Bucket
}
}
return musiclogic.NewService(musicmodel.NewRepo(db.Gorm()), objects, publicBase, bucket)
}
func wireClient(db *mysql.DB, users *userlogic.Service) *clientlogic.Service {
var repo clientlogic.Repository
if db != nil && db.Ready() {
repo = clientmodel.NewRepo(db.Gorm())
}
return clientlogic.NewService(repo, users, users)
}
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, mode authlogic.PairKeyMode) authlogic.PairStore {
if rdb != nil && rdb.Ready() {
return authlogic.NewRedisPairsMode(rdb.RDB(), prefix, mode)
}
return authlogic.NewMemoryPairsMode(mode)
}
func wireDevices(db *mysql.DB) *devicelogic.Service {
if db != nil && db.Ready() {
return devicelogic.NewService(devicemodel.NewRegistryRepo(db.Gorm()))
}
return devicelogic.NewService(devicemodel.NewMemoryRegistry())
}
func (a *App) Run() error {
if a.deliver != nil {
a.deliver.Start(context.Background())
}
if a.squareBot != nil {
a.squareBot.Start()
a.log.Info("squarebot started")
}
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()
}
if a.squareBot != nil {
a.squareBot.Stop()
}
err = errors.Join(err, a.redis.Close(), a.mysql.Close())
if a.logClose != nil {
err = errors.Join(err, a.logClose())
}
return err
}