diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..77c622e --- /dev/null +++ b/.dockerignore @@ -0,0 +1,6 @@ +.git +**/node_modules +**/dist +**/.run +**/*.log +npm-debug.log* diff --git a/backend/Dockerfile b/backend/Dockerfile index 4296ea8..de2ddc2 100644 --- a/backend/Dockerfile +++ b/backend/Dockerfile @@ -11,7 +11,7 @@ ARG GO_VERSION=1.24.5 ARG GOPROXY=https://goproxy.cn,direct ARG GOSUMDB=sum.golang.google.cn -FROM golang:${GO_VERSION} AS build +FROM golang:${GO_VERSION} AS deps ARG GOPROXY ARG GOSUMDB ENV GOPROXY=${GOPROXY} \ @@ -24,12 +24,17 @@ RUN --mount=type=cache,target=/go/pkg/mod \ --mount=type=cache,target=/root/.cache/go-build \ go mod download +FROM deps AS source COPY backend/ ./ ENV CGO_ENABLED=0 + +FROM source AS api-build RUN --mount=type=cache,target=/go/pkg/mod \ --mount=type=cache,target=/root/.cache/go-build \ go build -trimpath -ldflags="-s -w" -o /out/api ./cmd + +FROM source AS worker-build RUN --mount=type=cache,target=/go/pkg/mod \ --mount=type=cache,target=/root/.cache/go-build \ go build -trimpath -ldflags="-s -w" -o /out/worker ./cmd/worker @@ -37,15 +42,15 @@ RUN --mount=type=cache,target=/go/pkg/mod \ FROM alpine:3.21 AS base RUN apk add --no-cache ca-certificates tzdata && adduser -D -H -s /sbin/nologin app WORKDIR /app -COPY --from=build /src/backend/configs ./configs +COPY --from=source /src/backend/configs ./configs RUN mkdir -p ./.run/uploads && chown -R app:app /app USER app FROM base AS api -COPY --from=build /out/api /app/api +COPY --from=api-build /out/api /app/api EXPOSE 8080 ENTRYPOINT ["/app/api"] FROM base AS worker -COPY --from=build /out/worker /app/worker +COPY --from=worker-build /out/worker /app/worker ENTRYPOINT ["/app/worker"] diff --git a/backend/internal/account/service.go b/backend/internal/account/service.go index 4f1e85c..611820b 100644 --- a/backend/internal/account/service.go +++ b/backend/internal/account/service.go @@ -165,7 +165,9 @@ func (as *AccountService) Logout(ctx context.Context, accountID uint) error { log.Printf("failed to del refresh cache: %v", err) } if account.RefreshToken != "" { - as.cache.Del(cacheCtx, as.cache.Key("refresh:%s", account.RefreshToken)) + if err := as.cache.Del(cacheCtx, as.cache.Key("refresh:%s", account.RefreshToken)); err != nil { + log.Printf("failed to del refresh lookup: %v", err) + } } } return as.accountRepository.Logout(ctx, account.ID) @@ -211,8 +213,12 @@ func (as *AccountService) RefreshAccessToken(ctx context.Context, refreshToken s if err != nil { return "", 0, "", err } - as.accountRepository.UpdateToken(ctx, account.ID, newToken) - as.cache.SetBytes(cacheCtx, as.cache.Key("account:%d", account.ID), []byte(newToken), 24*time.Hour) + if err := as.accountRepository.UpdateToken(ctx, account.ID, newToken); err != nil { + return "", 0, "", err + } + if err := as.cache.SetBytes(cacheCtx, as.cache.Key("account:%d", account.ID), []byte(newToken), 24*time.Hour); err != nil { + log.Printf("failed to set cache: %v", err) + } return newToken, account.ID, account.Username, nil } } @@ -228,7 +234,9 @@ func (as *AccountService) RefreshAccessToken(ctx context.Context, refreshToken s if err != nil { return "", 0, "", err } - as.accountRepository.UpdateToken(ctx, acc.ID, newToken) + if err := as.accountRepository.UpdateToken(ctx, acc.ID, newToken); err != nil { + return "", 0, "", err + } return newToken, acc.ID, acc.Username, nil } } diff --git a/backend/internal/feed/service.go b/backend/internal/feed/service.go index f539c45..f8ec541 100644 --- a/backend/internal/feed/service.go +++ b/backend/internal/feed/service.go @@ -147,11 +147,15 @@ func (f *FeedService) GetVideoByIDs(ctx context.Context, videoIDs []uint) ([]*vi // 查询最新视频 (冷热分离 + 游标分页) func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore time.Time, viewerAccountID uint) (ListLatestResponse, error) { + if f.rediscache == nil { + return f.listLatestFromDB(ctx, limit, latestBefore, viewerAccountID) + } + // 获取 ZSET 中最老的一条数据 zsetTail, err := f.rediscache.ZRangeWithScores(ctx, f.rediscache.Key("feed:global_timeline"), 0, 0) if err != nil { - return ListLatestResponse{}, err + return f.listLatestFromDB(ctx, limit, latestBefore, viewerAccountID) } isZsetEmpty := len(zsetTail) == 0 @@ -271,9 +275,7 @@ func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore ti // 将本页最后一条视频的时间作为下一次请求的游标 nextTime = baseVideos[len(baseVideos)-1].CreateTime.UnixMilli() } - var hasMore bool - - hasMore = len(baseVideos) == limit + hasMore := len(baseVideos) == limit feedVideos, err := f.buildFeedVideos(ctx, baseVideos, viewerAccountID) if err != nil { @@ -287,6 +289,26 @@ func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore ti }, nil } +func (f *FeedService) listLatestFromDB(ctx context.Context, limit int, latestBefore time.Time, viewerAccountID uint) (ListLatestResponse, error) { + videos, err := f.repo.ListLatest(ctx, limit, latestBefore) + if err != nil { + return ListLatestResponse{}, err + } + feedVideos, err := f.buildFeedVideos(ctx, videos, viewerAccountID) + if err != nil { + return ListLatestResponse{}, err + } + var nextTime int64 + if len(videos) > 0 { + nextTime = videos[len(videos)-1].CreateTime.UnixMilli() + } + return ListLatestResponse{ + VideoList: feedVideos, + NextTime: nextTime, + HasMore: len(videos) == limit, + }, nil +} + // 按照点赞数查询视频 func (f *FeedService) ListLikesCount(ctx context.Context, limit int, cursor *LikesCountCursor, viewerAccountID uint) (ListLikesCountResponse, error) { videos, err := f.repo.ListLikesCountWithCursor(ctx, limit, cursor) diff --git a/backend/internal/http/router.go b/backend/internal/http/router.go index 46912e3..df6f420 100644 --- a/backend/internal/http/router.go +++ b/backend/internal/http/router.go @@ -12,10 +12,11 @@ import ( "feedsystem_video_go/internal/social" "feedsystem_video_go/internal/video" "feedsystem_video_go/internal/worker" - "github.com/gin-gonic/gin" - "gorm.io/gorm" "log" "time" + + "github.com/gin-gonic/gin" + "gorm.io/gorm" ) func SetRouter(db *gorm.DB, cache *rediscache.Client, rmq *rabbitmq.RabbitMQ) *gin.Engine { @@ -23,6 +24,9 @@ func SetRouter(db *gorm.DB, cache *rediscache.Client, rmq *rabbitmq.RabbitMQ) *g if err := r.SetTrustedProxies(nil); err != nil { log.Printf("SetTrustedProxies failed: %v", err) } + r.GET("/healthz", func(c *gin.Context) { + c.JSON(200, gin.H{"status": "ok"}) + }) r.Static("/static", "./.run/uploads") // rate_limit loginLimiter := ratelimit.Limit(cache, "account_login", 10, time.Minute, ratelimit.KeyByIP) @@ -201,9 +205,15 @@ func SetRouter(db *gorm.DB, cache *rediscache.Client, rmq *rabbitmq.RabbitMQ) *g // SSE notification if rmq != nil && rmq.Ch != nil { - rmq.DeclareTopic("like.events", "notification.like", "like.like") - rmq.DeclareTopic("comment.events", "notification.comment", "comment.publish") - rmq.DeclareTopic("social.events", "notification.social", "social.follow") + if err := rmq.DeclareTopic("like.events", "notification.like", "like.like"); err != nil { + log.Printf("notification like topic init failed: %v", err) + } + if err := rmq.DeclareTopic("comment.events", "notification.comment", "comment.publish"); err != nil { + log.Printf("notification comment topic init failed: %v", err) + } + if err := rmq.DeclareTopic("social.events", "notification.social", "social.follow"); err != nil { + log.Printf("notification social topic init failed: %v", err) + } } sseHub := worker.NewSSEHub(db) notifGroup := r.Group("/notification") @@ -216,19 +226,37 @@ func SetRouter(db *gorm.DB, cache *rediscache.Client, rmq *rabbitmq.RabbitMQ) *g ctx := context.Background() // consume from like queue go func() { - w := worker.NewNotificationWorker(rmq.Ch, db, "notification.like", hub) + ch, err := rmq.Conn.Channel() + if err != nil { + log.Printf("notification-like channel: %v", err) + return + } + defer ch.Close() + w := worker.NewNotificationWorker(ch, db, "notification.like", hub) if err := w.Run(ctx); err != nil { log.Printf("notification-like worker: %v", err) } }() go func() { - w := worker.NewNotificationWorker(rmq.Ch, db, "notification.comment", hub) + ch, err := rmq.Conn.Channel() + if err != nil { + log.Printf("notification-comment channel: %v", err) + return + } + defer ch.Close() + w := worker.NewNotificationWorker(ch, db, "notification.comment", hub) if err := w.Run(ctx); err != nil { log.Printf("notification-comment worker: %v", err) } }() go func() { - w := worker.NewNotificationWorker(rmq.Ch, db, "notification.social", hub) + ch, err := rmq.Conn.Channel() + if err != nil { + log.Printf("notification-social channel: %v", err) + return + } + defer ch.Close() + w := worker.NewNotificationWorker(ch, db, "notification.social", hub) if err := w.Run(ctx); err != nil { log.Printf("notification-social worker: %v", err) } diff --git a/backend/internal/middleware/rabbitmq/rabbitMQ.go b/backend/internal/middleware/rabbitmq/rabbitMQ.go index b665051..1b8eeda 100644 --- a/backend/internal/middleware/rabbitmq/rabbitMQ.go +++ b/backend/internal/middleware/rabbitmq/rabbitMQ.go @@ -30,22 +30,28 @@ func NewRabbitMQ(cfg *config.RabbitMQConfig) (*RabbitMQ, error) { } ch, err := conn.Channel() if err != nil { + _ = conn.Close() return nil, err } return &RabbitMQ{Conn: conn, Ch: ch}, nil } func (r *RabbitMQ) Close() error { - if r == nil || r.Ch == nil || r.Conn == nil { + if r == nil { return nil } - if err := r.Ch.Close(); err != nil { - return err + var closeErr error + if r.Ch != nil { + if err := r.Ch.Close(); err != nil { + closeErr = err + } } - if err := r.Conn.Close(); err != nil { - return err + if r.Conn != nil { + if err := r.Conn.Close(); closeErr == nil && err != nil { + closeErr = err + } } - return nil + return closeErr } func (r *RabbitMQ) DeclareTopic(exchange string, queue string, bindingKey string) error { diff --git a/backend/internal/middleware/ratelimit/ratelimit.go b/backend/internal/middleware/ratelimit/ratelimit.go index 700b978..0f38d44 100644 --- a/backend/internal/middleware/ratelimit/ratelimit.go +++ b/backend/internal/middleware/ratelimit/ratelimit.go @@ -4,11 +4,12 @@ import ( jwt "feedsystem_video_go/internal/middleware/jwt" rediscache "feedsystem_video_go/internal/middleware/redis" "fmt" - "github.com/gin-gonic/gin" "net/http" "strconv" "strings" "time" + + "github.com/gin-gonic/gin" ) type KeyFunc func(*gin.Context) (string, bool) diff --git a/backend/internal/middleware/redis/cache.go b/backend/internal/middleware/redis/cache.go index 270cda3..b3e328c 100644 --- a/backend/internal/middleware/redis/cache.go +++ b/backend/internal/middleware/redis/cache.go @@ -2,21 +2,34 @@ package redis import ( "context" + "errors" "time" ) func (c *Client) GetBytes(ctx context.Context, key string) ([]byte, error) { + if c == nil || c.rdb == nil { + return nil, errors.New("redis client not initialized") + } return c.rdb.Get(ctx, key).Bytes() } func (c *Client) SetBytes(ctx context.Context, key string, value []byte, ttl time.Duration) error { + if c == nil || c.rdb == nil { + return errors.New("redis client not initialized") + } return c.rdb.Set(ctx, key, value, ttl).Err() } func (c *Client) Del(ctx context.Context, key string) error { + if c == nil || c.rdb == nil { + return errors.New("redis client not initialized") + } return c.rdb.Del(ctx, key).Err() } func (c *Client) MGet(cacheCtx context.Context, cacheKeys ...string) ([]interface{}, error) { + if c == nil || c.rdb == nil { + return nil, errors.New("redis client not initialized") + } return c.rdb.MGet(cacheCtx, cacheKeys...).Result() } diff --git a/backend/internal/middleware/redis/redis.go b/backend/internal/middleware/redis/redis.go index 19f9697..9e10185 100644 --- a/backend/internal/middleware/redis/redis.go +++ b/backend/internal/middleware/redis/redis.go @@ -4,6 +4,7 @@ import ( "context" "crypto/rand" "encoding/hex" + "errors" "feedsystem_video_go/internal/config" "fmt" "strconv" @@ -41,7 +42,7 @@ func (c *Client) Close() error { func (c *Client) Ping(ctx context.Context) error { if c == nil || c.rdb == nil { - return nil + return errors.New("redis client not initialized") } return c.rdb.Ping(ctx).Err() } diff --git a/backend/internal/video/chunk_handler.go b/backend/internal/video/chunk_handler.go index b7cf753..30f3ec5 100644 --- a/backend/internal/video/chunk_handler.go +++ b/backend/internal/video/chunk_handler.go @@ -3,6 +3,7 @@ package video import ( "crypto/md5" "encoding/json" + "errors" "fmt" "io" "net/http" @@ -18,6 +19,8 @@ import ( const sessionTTL = 24 * time.Hour +var errChunkCacheUnavailable = errors.New("chunk upload requires redis") + type ChunkUploadHandler struct { cache *rediscache.Client } @@ -35,6 +38,9 @@ func (h *ChunkUploadHandler) hashKey(accountID uint, fileHash string) string { } func (h *ChunkUploadHandler) getSession(ctx *gin.Context, uploadID string) (*ChunkUploadSession, error) { + if h.cache == nil { + return nil, errChunkCacheUnavailable + } b, err := h.cache.GetBytes(ctx.Request.Context(), h.sessionKey(uploadID)) if err != nil { return nil, fmt.Errorf("upload session not found") @@ -47,6 +53,9 @@ func (h *ChunkUploadHandler) getSession(ctx *gin.Context, uploadID string) (*Chu } func (h *ChunkUploadHandler) saveSession(ctx *gin.Context, s *ChunkUploadSession) error { + if h.cache == nil { + return errChunkCacheUnavailable + } b, err := json.Marshal(s) if err != nil { return err @@ -55,6 +64,11 @@ func (h *ChunkUploadHandler) saveSession(ctx *gin.Context, s *ChunkUploadSession } func (h *ChunkUploadHandler) InitChunkUpload(c *gin.Context) { + if h.cache == nil { + c.JSON(http.StatusServiceUnavailable, gin.H{"error": errChunkCacheUnavailable.Error()}) + return + } + var req InitChunkUploadRequest if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) @@ -120,6 +134,11 @@ func (h *ChunkUploadHandler) InitChunkUpload(c *gin.Context) { } func (h *ChunkUploadHandler) UploadChunk(c *gin.Context) { + if h.cache == nil { + c.JSON(http.StatusServiceUnavailable, gin.H{"error": errChunkCacheUnavailable.Error()}) + return + } + var req UploadChunkRequest if err := c.ShouldBind(&req); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) @@ -211,6 +230,11 @@ func (h *ChunkUploadHandler) UploadChunk(c *gin.Context) { } func (h *ChunkUploadHandler) ChunkStatus(c *gin.Context) { + if h.cache == nil { + c.JSON(http.StatusServiceUnavailable, gin.H{"error": errChunkCacheUnavailable.Error()}) + return + } + var req ChunkStatusRequest if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) @@ -241,6 +265,11 @@ func (h *ChunkUploadHandler) ChunkStatus(c *gin.Context) { } func (h *ChunkUploadHandler) CompleteChunkUpload(c *gin.Context) { + if h.cache == nil { + c.JSON(http.StatusServiceUnavailable, gin.H{"error": errChunkCacheUnavailable.Error()}) + return + } + var req CompleteChunkUploadRequest if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) @@ -328,11 +357,11 @@ func (h *ChunkUploadHandler) CompleteChunkUpload(c *gin.Context) { finalFile.Close() // Clean up temp chunks - os.RemoveAll(tmpDir) + _ = os.RemoveAll(tmpDir) // Clean up Redis session - h.cache.Del(c.Request.Context(), h.sessionKey(req.UploadID)) - h.cache.Del(c.Request.Context(), h.hashKey(accountID, session.FileHash)) + _ = h.cache.Del(c.Request.Context(), h.sessionKey(req.UploadID)) + _ = h.cache.Del(c.Request.Context(), h.hashKey(accountID, session.FileHash)) urlPath := fmt.Sprintf("/static/videos/%d/%s/%s.mp4", accountID, date, filename) playURL := buildAbsoluteURL(c, urlPath) diff --git a/backend/internal/video/comment_service.go b/backend/internal/video/comment_service.go index 594be4b..e115ee7 100644 --- a/backend/internal/video/comment_service.go +++ b/backend/internal/video/comment_service.go @@ -6,6 +6,7 @@ import ( "feedsystem_video_go/internal/apierror" "feedsystem_video_go/internal/middleware/rabbitmq" rediscache "feedsystem_video_go/internal/middleware/redis" + "log" "regexp" "strings" @@ -150,6 +151,8 @@ func (s *CommentService) notifyMentions(ctx context.Context, comment *Comment) { TargetID: comment.VideoID, Content: comment.Username + " 在评论中提到了你", } - s.repo.db.WithContext(ctx).Table("notifications").Create(¬if) + if err := s.repo.db.WithContext(ctx).Table("notifications").Create(¬if).Error; err != nil { + log.Printf("create mention notification failed: %v", err) + } } } diff --git a/backend/internal/video/video_service.go b/backend/internal/video/video_service.go index e387695..f2ad828 100644 --- a/backend/internal/video/video_service.go +++ b/backend/internal/video/video_service.go @@ -64,8 +64,12 @@ func (vs *VideoService) Publish(ctx context.Context, video *Video) error { tags := ExtractTags(video.Title + " " + video.Description) for _, tagName := range tags { var tag Tag - tx.Where("name = ?", tagName).FirstOrCreate(&tag, Tag{Name: tagName}) - tx.Create(&VideoTag{VideoID: video.ID, TagID: tag.ID}) + if err := tx.Where("name = ?", tagName).FirstOrCreate(&tag, Tag{Name: tagName}).Error; err != nil { + return err + } + if err := tx.Create(&VideoTag{VideoID: video.ID, TagID: tag.ID}).Error; err != nil { + return err + } } return nil }) diff --git a/backend/internal/worker/likeworker.go b/backend/internal/worker/likeworker.go index dfe8826..517cc37 100644 --- a/backend/internal/worker/likeworker.go +++ b/backend/internal/worker/likeworker.go @@ -6,9 +6,10 @@ import ( "errors" "feedsystem_video_go/internal/middleware/rabbitmq" "feedsystem_video_go/internal/video" - amqp "github.com/rabbitmq/amqp091-go" "log" "time" + + amqp "github.com/rabbitmq/amqp091-go" ) type LikeWorker struct { diff --git a/backend/internal/worker/notificationworker.go b/backend/internal/worker/notificationworker.go index 2072f56..98fba5a 100644 --- a/backend/internal/worker/notificationworker.go +++ b/backend/internal/worker/notificationworker.go @@ -42,6 +42,9 @@ func (w *NotificationWorker) Run(ctx context.Context) error { if w == nil || w.ch == nil || w.db == nil { return errors.New("notification worker is not initialized") } + if w.queue == "" { + return errors.New("queue is required") + } if err := w.db.WithContext(ctx).AutoMigrate(&Notification{}); err != nil { return err } @@ -96,10 +99,12 @@ func (w *NotificationWorker) process(ctx context.Context, d amqp.Delivery) error return nil } var authorID uint - w.db.WithContext(ctx).Model(&struct { + if err := w.db.WithContext(ctx).Model(&struct { ID uint AuthorID uint - }{}).Table("videos").Where("id = ?", evt.VideoID).Select("author_id").Scan(&authorID) + }{}).Table("videos").Where("id = ?", evt.VideoID).Select("author_id").Scan(&authorID).Error; err != nil { + return err + } if authorID == 0 || authorID == evt.UserID { return nil } @@ -114,10 +119,12 @@ func (w *NotificationWorker) process(ctx context.Context, d amqp.Delivery) error return nil } var authorID uint - w.db.WithContext(ctx).Model(&struct { + if err := w.db.WithContext(ctx).Model(&struct { ID uint AuthorID uint - }{}).Table("videos").Where("id = ?", evt.VideoID).Select("author_id").Scan(&authorID) + }{}).Table("videos").Where("id = ?", evt.VideoID).Select("author_id").Scan(&authorID).Error; err != nil { + return err + } if authorID == 0 || authorID == evt.AuthorID { return nil } diff --git a/backend/internal/worker/outboxworker.go b/backend/internal/worker/outboxworker.go index 6bc27eb..1e8ced0 100644 --- a/backend/internal/worker/outboxworker.go +++ b/backend/internal/worker/outboxworker.go @@ -35,7 +35,9 @@ func StartOutboxPoller(db *gorm.DB, tmq *rabbitmq.TimelineMQ) { err := tmq.PublishVideo(context.Background(), msg.VideoID, msg.CreateTime) if err == nil { - db.Delete(&msg) + if err := db.Delete(&msg).Error; err != nil { + log.Printf("删除 outbox 消息失败: id=%d, err=%v", msg.ID, err) + } } else { log.Printf("投递MQ失败: VideoID: %d, err: %v", msg.VideoID, err) } diff --git a/backend/internal/worker/ssehub.go b/backend/internal/worker/ssehub.go index ca855d9..d5a8862 100644 --- a/backend/internal/worker/ssehub.go +++ b/backend/internal/worker/ssehub.go @@ -2,7 +2,9 @@ package worker import ( "encoding/json" + "errors" "fmt" + "io" "net/http" "sync" "time" @@ -25,8 +27,8 @@ func NewSSEHub(db *gorm.DB) *SSEHub { func (h *SSEHub) Push(userID uint, n *Notification) { h.mu.RLock() + defer h.mu.RUnlock() chs, ok := h.clients[userID] - h.mu.RUnlock() if !ok { return } @@ -52,13 +54,27 @@ func (h *SSEHub) Unsubscribe(userID uint, ch chan *Notification) { chs := h.clients[userID] for i, c := range chs { if c == ch { - h.clients[userID] = append(chs[:i], chs[i+1:]...) + chs = append(chs[:i], chs[i+1:]...) + if len(chs) == 0 { + delete(h.clients, userID) + } else { + h.clients[userID] = chs + } close(c) return } } } +func sseAccountID(c *gin.Context) (uint, bool) { + accountID, ok := c.Get("accountID") + if !ok { + return 0, false + } + userID, ok := accountID.(uint) + return userID, ok && userID != 0 +} + func (h *SSEHub) SSERequireAuth() gin.HandlerFunc { return func(c *gin.Context) { token := c.Query("token") @@ -83,8 +99,11 @@ func (h *SSEHub) SSERequireAuth() gin.HandlerFunc { } func (h *SSEHub) SSEHandler(c *gin.Context) { - accountID, _ := c.Get("accountID") - userID := accountID.(uint) + userID, ok := sseAccountID(c) + if !ok { + c.JSON(http.StatusUnauthorized, gin.H{"error": "invalid account"}) + return + } c.Writer.Header().Set("Content-Type", "text/event-stream") c.Writer.Header().Set("Cache-Control", "no-cache") @@ -120,8 +139,11 @@ func (h *SSEHub) SSEHandler(c *gin.Context) { } func (h *SSEHub) ListHandler(c *gin.Context) { - accountID, _ := c.Get("accountID") - userID := accountID.(uint) + userID, ok := sseAccountID(c) + if !ok { + c.JSON(http.StatusUnauthorized, gin.H{"error": "invalid account"}) + return + } var notifications []Notification if err := h.db.WithContext(c.Request.Context()). @@ -139,28 +161,45 @@ func (h *SSEHub) ListHandler(c *gin.Context) { } func (h *SSEHub) MarkReadHandler(c *gin.Context) { - accountID, _ := c.Get("accountID") - userID := accountID.(uint) + userID, ok := sseAccountID(c) + if !ok { + c.JSON(http.StatusUnauthorized, gin.H{"error": "invalid account"}) + return + } var req struct { ID *uint `json:"id"` } - c.ShouldBindJSON(&req) + if err := c.ShouldBindJSON(&req); err != nil && !errors.Is(err, io.EOF) { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + var err error if req.ID != nil { - h.db.WithContext(c.Request.Context()).Model(&Notification{}).Where("id = ? AND recipient_id = ?", *req.ID, userID).Update("is_read", true) + err = h.db.WithContext(c.Request.Context()).Model(&Notification{}).Where("id = ? AND recipient_id = ?", *req.ID, userID).Update("is_read", true).Error } else { - h.db.WithContext(c.Request.Context()).Model(&Notification{}).Where("recipient_id = ?", userID).Update("is_read", true) + err = h.db.WithContext(c.Request.Context()).Model(&Notification{}).Where("recipient_id = ?", userID).Update("is_read", true).Error + } + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return } c.JSON(200, gin.H{"message": "ok"}) } func (h *SSEHub) UnreadCountHandler(c *gin.Context) { - accountID, _ := c.Get("accountID") - userID := accountID.(uint) + userID, ok := sseAccountID(c) + if !ok { + c.JSON(http.StatusUnauthorized, gin.H{"error": "invalid account"}) + return + } var count int64 - h.db.WithContext(c.Request.Context()).Model(&Notification{}).Where("recipient_id = ? AND is_read = ?", userID, false).Count(&count) + if err := h.db.WithContext(c.Request.Context()).Model(&Notification{}).Where("recipient_id = ? AND is_read = ?", userID, false).Count(&count).Error; err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } c.JSON(200, gin.H{"count": count}) } diff --git a/docker-compose.yml b/docker-compose.yml index 26a7553..e7eae6a 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -60,6 +60,7 @@ services: target: api restart: always environment: + CONFIG_PATH: /app/configs/config.yaml JWT_SECRET: ${JWT_SECRET:-feedsystem-dev-secret-key} MYSQL_DATABASE: ${MYSQL_DATABASE:-feedsystem} MYSQL_ROOT_PASSWORD: ${MYSQL_ROOT_PASSWORD:-123456} @@ -79,7 +80,7 @@ services: rabbitmq: condition: service_healthy healthcheck: - test: ["CMD-SHELL", "pgrep api || exit 1"] + test: ["CMD-SHELL", "wget -qO- http://127.0.0.1:8080/healthz || exit 1"] interval: 10s timeout: 5s retries: 3 @@ -91,6 +92,7 @@ services: target: worker restart: always environment: + CONFIG_PATH: /app/configs/config.yaml JWT_SECRET: ${JWT_SECRET:-feedsystem-dev-secret-key} MYSQL_DATABASE: ${MYSQL_DATABASE:-feedsystem} MYSQL_ROOT_PASSWORD: ${MYSQL_ROOT_PASSWORD:-123456} diff --git a/frontend/src/stores/auth.ts b/frontend/src/stores/auth.ts index 5d9fc32..ad734d9 100644 --- a/frontend/src/stores/auth.ts +++ b/frontend/src/stores/auth.ts @@ -11,11 +11,11 @@ function readStored(key: string): string | null { } function writeStored(key: string, value: string) { - localStorage.setItem(key, value) + try { localStorage.setItem(key, value) } catch {} } function removeStored(key: string) { - localStorage.removeItem(key) + try { localStorage.removeItem(key) } catch {} } export const useAuthStore = defineStore('auth', () => {