438 lines
13 KiB
Go
438 lines
13 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"
|
|
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"
|
|
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()}
|
|
users := wireUsers(idgen, db, rdb, 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)
|
|
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, &httpx.EngineDeps{
|
|
Auth: jwtSvc,
|
|
Login: login,
|
|
Users: users,
|
|
Friends: friends,
|
|
Groups: groups,
|
|
Files: files,
|
|
Favorites: favorites,
|
|
Messages: messages,
|
|
Sync: syncSvc,
|
|
Health: probe,
|
|
AuthLimiter: authLimit,
|
|
})
|
|
return app, nil
|
|
}
|
|
|
|
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 wireUsers(idgen *id.Generator, db *mysql.DB, rdb *redisx.Client, store *objectstore.Store) *userlogic.Service {
|
|
var repo userlogic.Repository
|
|
if db != nil && db.Ready() {
|
|
repo = usermodel.NewRepo(db.Gorm())
|
|
} else {
|
|
repo = usermodel.NewMemoryRepo()
|
|
}
|
|
var codes userlogic.CodeStore
|
|
if rdb != nil && rdb.Ready() {
|
|
codes = usermodel.NewRedisCodes(rdb.RDB())
|
|
} else {
|
|
codes = usermodel.NewMemoryCodes()
|
|
}
|
|
var objects userlogic.Objects
|
|
if store != nil && store.Ready() {
|
|
objects = store
|
|
}
|
|
return userlogic.NewService(repo, codes, idgen, objects)
|
|
}
|
|
|
|
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 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 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
|
|
}
|