From ee78071d4dca8abadc706f4b623d166161e2ffce Mon Sep 17 00:00:00 2001 From: lan Date: Mon, 4 May 2026 13:07:03 +0800 Subject: [PATCH] refactor: improve system stability, performance, and code structure This commit introduces several architectural improvements and optimizations across the codebase: - **Performance & Reliability**: - Implemented Redis pipelining in `ConversationCache.CacheMessage` to reduce network round-trips. - Added a circuit breaker to the JPush client to prevent cascading failures. - Introduced batch deletion and batch member addition capabilities in repositories. - Added message idempotency support using `client_msg_id` and a Redis-based cache. - Optimized WebSocket handling with connection limits (total and per-user) and improved error logging. - **Code Refactoring**: - Refactored `Router` to use a `RouterDeps` struct, simplifying the constructor and improving maintainability. - Unified model ID generation logic using new `id_helper.go` (supporting UUID and Snowflake). - Standardized JSON serialization/deserialization in models using `json_helper.go`. - Refactored DTO conversion logic, specifically for `UserResponse` (using functional options) and `Report` responses. - Removed redundant/deprecated DTOs like `PostDetailResponse` and `TradeItemDetailResponse`. - **Cache Improvements**: - Enhanced `LayeredCache` with `SetRaw` to avoid double-encoding when promoting values from Redis to local cache. - Added `DeleteBatch` support to the cache interface. - **Other Changes**: - Cleaned up `config.go` by removing redundant default values and explicit environment variable overrides. - Improved WebSocket registration flow to handle connection limits gracefully. --- .kilo/package-lock.json | 281 +++++++++++++++++- cmd/server/app.go | 18 +- cmd/server/wire.go | 76 ++--- cmd/server/wire_gen.go | 78 ++--- internal/cache/cache.go | 2 + internal/cache/conversation_cache.go | 41 ++- internal/cache/keys.go | 21 +- internal/cache/layered_cache.go | 23 +- internal/cache/lru.go | 33 ++ internal/cache/lru_striped.go | 13 +- internal/cache/redis_cache.go | 25 +- internal/config/config.go | 107 +------ internal/dto/dto.go | 41 +-- internal/dto/post_converter.go | 45 +-- internal/dto/report.go | 58 ++-- internal/dto/trade_converter.go | 34 +-- internal/dto/trade_dto.go | 20 -- internal/dto/user_converter.go | 220 +++----------- internal/handler/message_handler.go | 5 +- internal/handler/post_handler.go | 2 +- internal/handler/user_handler.go | 2 +- internal/handler/ws_handler.go | 25 +- internal/model/audit_log.go | 5 +- internal/model/call.go | 12 +- internal/model/channel.go | 5 +- internal/model/comment.go | 9 +- internal/model/conversation.go | 12 +- internal/model/device_token.go | 11 +- internal/model/favorite.go | 5 +- internal/model/follow.go | 5 +- internal/model/group.go | 11 +- internal/model/group_announcement.go | 12 +- internal/model/group_join_request.go | 12 +- internal/model/group_member.go | 12 +- internal/model/id_helper.go | 56 ++++ internal/model/json_helper.go | 28 ++ internal/model/like.go | 5 +- internal/model/material.go | 23 +- internal/model/message.go | 40 +-- internal/model/message_read.go | 19 -- internal/model/notification.go | 5 +- internal/model/operation_log.go | 5 +- internal/model/post.go | 9 +- internal/model/post_reference.go | 5 +- internal/model/push_record.go | 11 +- internal/model/report.go | 5 +- internal/model/schedule_course.go | 5 +- internal/model/sensitive_word.go | 5 +- internal/model/sticker.go | 5 +- internal/model/system_notification.go | 29 +- internal/model/trade.go | 13 +- internal/model/user.go | 5 +- internal/model/user_block.go | 5 +- internal/model/user_profile_audit.go | 5 +- internal/model/verification.go | 5 +- internal/model/vote.go | 9 +- internal/pkg/circuitbreaker/circuitbreaker.go | 136 +++++++++ internal/pkg/jpush/client.go | 19 +- internal/pkg/response/response.go | 7 - internal/pkg/ws/hub.go | 225 ++++++++++++-- .../repository/group_join_request_repo.go | 25 ++ internal/repository/group_repo.go | 18 +- internal/router/router.go | 155 +++++----- internal/service/chat_service.go | 99 ++++-- internal/service/trade_service.go | 6 +- 65 files changed, 1293 insertions(+), 975 deletions(-) create mode 100644 internal/model/id_helper.go create mode 100644 internal/model/json_helper.go delete mode 100644 internal/model/message_read.go create mode 100644 internal/pkg/circuitbreaker/circuitbreaker.go diff --git a/.kilo/package-lock.json b/.kilo/package-lock.json index 86ea34f..4240bb2 100644 --- a/.kilo/package-lock.json +++ b/.kilo/package-lock.json @@ -5,21 +5,22 @@ "packages": { "": { "dependencies": { - "@kilocode/plugin": "7.2.10" + "@kilocode/plugin": "7.2.31" } }, "node_modules/@kilocode/plugin": { - "version": "7.2.10", - "resolved": "https://registry.npmjs.org/@kilocode/plugin/-/plugin-7.2.10.tgz", - "integrity": "sha512-VJPhJC+E5WWu7XgEJzrVOxKJlwJ+OATwxEzgjqEPj8KN5N38YxUPBY/rzUTjv90x7nkzyk1rFGfCVqXdA/Koug==", + "version": "7.2.31", + "resolved": "https://registry.npmjs.org/@kilocode/plugin/-/plugin-7.2.31.tgz", + "integrity": "sha512-KmKTTIly7hRlJdXKhqZ/j/brvTPh0z0UTjWSjJWq5fqf4pATgYGn7G0g3ZjILnN7MUkkZXuljgqExTEeQJHGkQ==", "license": "MIT", "dependencies": { - "@kilocode/sdk": "7.2.10", + "@kilocode/sdk": "7.2.31", + "effect": "4.0.0-beta.48", "zod": "4.1.8" }, "peerDependencies": { - "@opentui/core": ">=0.1.97", - "@opentui/solid": ">=0.1.97" + "@opentui/core": ">=0.1.100", + "@opentui/solid": ">=0.1.100" }, "peerDependenciesMeta": { "@opentui/core": { @@ -31,14 +32,98 @@ } }, "node_modules/@kilocode/sdk": { - "version": "7.2.10", - "resolved": "https://registry.npmjs.org/@kilocode/sdk/-/sdk-7.2.10.tgz", - "integrity": "sha512-H6jGXYAhN/yjOGX3MRZ0OxyEAuRGY3VOwDbLTh4O6ljpgutFHaLvomDZ82qNVy7gl7AjJgi3SAQAt9UQpeGl/w==", + "version": "7.2.31", + "resolved": "https://registry.npmjs.org/@kilocode/sdk/-/sdk-7.2.31.tgz", + "integrity": "sha512-Sx05yv+3TIlc6M4Ze+YGgKCLoIg8B0WRE15JXDriVncT+wz7M6+e+4mjNWkwfsuywdeOTYPfeFViqX8iO7ckKA==", "license": "MIT", "dependencies": { "cross-spawn": "7.0.6" } }, + "node_modules/@msgpackr-extract/msgpackr-extract-darwin-arm64": { + "version": "3.0.3", + "resolved": "https://registry.npmjs.org/@msgpackr-extract/msgpackr-extract-darwin-arm64/-/msgpackr-extract-darwin-arm64-3.0.3.tgz", + "integrity": "sha512-QZHtlVgbAdy2zAqNA9Gu1UpIuI8Xvsd1v8ic6B2pZmeFnFcMWiPLfWXh7TVw4eGEZ/C9TH281KwhVoeQUKbyjw==", + "cpu": [ + "arm64" + ], + "license": "MIT", + "optional": true, + "os": [ + "darwin" + ] + }, + "node_modules/@msgpackr-extract/msgpackr-extract-darwin-x64": { + "version": "3.0.3", + "resolved": "https://registry.npmjs.org/@msgpackr-extract/msgpackr-extract-darwin-x64/-/msgpackr-extract-darwin-x64-3.0.3.tgz", + "integrity": "sha512-mdzd3AVzYKuUmiWOQ8GNhl64/IoFGol569zNRdkLReh6LRLHOXxU4U8eq0JwaD8iFHdVGqSy4IjFL4reoWCDFw==", + "cpu": [ + "x64" + ], + "license": "MIT", + "optional": true, + "os": [ + "darwin" + ] + }, + "node_modules/@msgpackr-extract/msgpackr-extract-linux-arm": { + "version": "3.0.3", + "resolved": "https://registry.npmjs.org/@msgpackr-extract/msgpackr-extract-linux-arm/-/msgpackr-extract-linux-arm-3.0.3.tgz", + "integrity": "sha512-fg0uy/dG/nZEXfYilKoRe7yALaNmHoYeIoJuJ7KJ+YyU2bvY8vPv27f7UKhGRpY6euFYqEVhxCFZgAUNQBM3nw==", + "cpu": [ + "arm" + ], + "license": "MIT", + "optional": true, + "os": [ + "linux" + ] + }, + "node_modules/@msgpackr-extract/msgpackr-extract-linux-arm64": { + "version": "3.0.3", + "resolved": "https://registry.npmjs.org/@msgpackr-extract/msgpackr-extract-linux-arm64/-/msgpackr-extract-linux-arm64-3.0.3.tgz", + "integrity": "sha512-YxQL+ax0XqBJDZiKimS2XQaf+2wDGVa1enVRGzEvLLVFeqa5kx2bWbtcSXgsxjQB7nRqqIGFIcLteF/sHeVtQg==", + "cpu": [ + "arm64" + ], + "license": "MIT", + "optional": true, + "os": [ + "linux" + ] + }, + "node_modules/@msgpackr-extract/msgpackr-extract-linux-x64": { + "version": "3.0.3", + "resolved": "https://registry.npmjs.org/@msgpackr-extract/msgpackr-extract-linux-x64/-/msgpackr-extract-linux-x64-3.0.3.tgz", + "integrity": "sha512-cvwNfbP07pKUfq1uH+S6KJ7dT9K8WOE4ZiAcsrSes+UY55E/0jLYc+vq+DO7jlmqRb5zAggExKm0H7O/CBaesg==", + "cpu": [ + "x64" + ], + "license": "MIT", + "optional": true, + "os": [ + "linux" + ] + }, + "node_modules/@msgpackr-extract/msgpackr-extract-win32-x64": { + "version": "3.0.3", + "resolved": "https://registry.npmjs.org/@msgpackr-extract/msgpackr-extract-win32-x64/-/msgpackr-extract-win32-x64-3.0.3.tgz", + "integrity": "sha512-x0fWaQtYp4E6sktbsdAqnehxDgEc/VwM7uLsRCYWaiGu0ykYdZPiS8zCWdnjHwyiumousxfBm4SO31eXqwEZhQ==", + "cpu": [ + "x64" + ], + "license": "MIT", + "optional": true, + "os": [ + "win32" + ] + }, + "node_modules/@standard-schema/spec": { + "version": "1.1.0", + "resolved": "https://registry.npmjs.org/@standard-schema/spec/-/spec-1.1.0.tgz", + "integrity": "sha512-l2aFy5jALhniG5HgqrD6jXLi/rUWrKvqN/qJx6yoJsgKhblVd+iqqU4RCXavm/jPityDo5TCvKMnpjKnOriy0w==", + "license": "MIT" + }, "node_modules/cross-spawn": { "version": "7.0.6", "resolved": "https://registry.npmjs.org/cross-spawn/-/cross-spawn-7.0.6.tgz", @@ -53,12 +138,135 @@ "node": ">= 8" } }, + "node_modules/detect-libc": { + "version": "2.1.2", + "resolved": "https://registry.npmjs.org/detect-libc/-/detect-libc-2.1.2.tgz", + "integrity": "sha512-Btj2BOOO83o3WyH59e8MgXsxEQVcarkUOpEYrubB0urwnN10yQ364rsiByU11nZlqWYZm05i/of7io4mzihBtQ==", + "license": "Apache-2.0", + "optional": true, + "engines": { + "node": ">=8" + } + }, + "node_modules/effect": { + "version": "4.0.0-beta.48", + "resolved": "https://registry.npmjs.org/effect/-/effect-4.0.0-beta.48.tgz", + "integrity": "sha512-MMAM/ZabuNdNmgXiin+BAanQXK7qM8mlt7nfXDoJ/Gn9V8i89JlCq+2N0AiWmqFLXjGLA0u3FjiOjSOYQk5uMw==", + "license": "MIT", + "dependencies": { + "@standard-schema/spec": "^1.1.0", + "fast-check": "^4.6.0", + "find-my-way-ts": "^0.1.6", + "ini": "^6.0.0", + "kubernetes-types": "^1.30.0", + "msgpackr": "^1.11.9", + "multipasta": "^0.2.7", + "toml": "^4.1.1", + "uuid": "^13.0.0", + "yaml": "^2.8.3" + } + }, + "node_modules/fast-check": { + "version": "4.7.0", + "resolved": "https://registry.npmjs.org/fast-check/-/fast-check-4.7.0.tgz", + "integrity": "sha512-NsZRtqvSSoCP0HbNjUD+r1JH8zqZalyp6gLY9e7OYs7NK9b6AHOs2baBFeBG7bVNsuoukh89x2Yg3rPsul8ziQ==", + "funding": [ + { + "type": "individual", + "url": "https://github.com/sponsors/dubzzz" + }, + { + "type": "opencollective", + "url": "https://opencollective.com/fast-check" + } + ], + "license": "MIT", + "dependencies": { + "pure-rand": "^8.0.0" + }, + "engines": { + "node": ">=12.17.0" + } + }, + "node_modules/find-my-way-ts": { + "version": "0.1.6", + "resolved": "https://registry.npmjs.org/find-my-way-ts/-/find-my-way-ts-0.1.6.tgz", + "integrity": "sha512-a85L9ZoXtNAey3Y6Z+eBWW658kO/MwR7zIafkIUPUMf3isZG0NCs2pjW2wtjxAKuJPxMAsHUIP4ZPGv0o5gyTA==", + "license": "MIT" + }, + "node_modules/ini": { + "version": "6.0.0", + "resolved": "https://registry.npmjs.org/ini/-/ini-6.0.0.tgz", + "integrity": "sha512-IBTdIkzZNOpqm7q3dRqJvMaldXjDHWkEDfrwGEQTs5eaQMWV+djAhR+wahyNNMAa+qpbDUhBMVt4ZKNwpPm7xQ==", + "license": "ISC", + "engines": { + "node": "^20.17.0 || >=22.9.0" + } + }, "node_modules/isexe": { "version": "2.0.0", "resolved": "https://registry.npmjs.org/isexe/-/isexe-2.0.0.tgz", "integrity": "sha512-RHxMLp9lnKHGHRng9QFhRCMbYAcVpn69smSGcq3f36xjgVVWThj4qqLbTLlq7Ssj8B+fIQ1EuCEGI2lKsyQeIw==", "license": "ISC" }, + "node_modules/kubernetes-types": { + "version": "1.30.0", + "resolved": "https://registry.npmjs.org/kubernetes-types/-/kubernetes-types-1.30.0.tgz", + "integrity": "sha512-Dew1okvhM/SQcIa2rcgujNndZwU8VnSapDgdxlYoB84ZlpAD43U6KLAFqYo17ykSFGHNPrg0qry0bP+GJd9v7Q==", + "license": "Apache-2.0" + }, + "node_modules/msgpackr": { + "version": "1.11.12", + "resolved": "https://registry.npmjs.org/msgpackr/-/msgpackr-1.11.12.tgz", + "integrity": "sha512-RBdJ1Un7yGlXWajrkxcSa93nvQ0w4zBf60c0yYv7YtBelP8H2FA7XsfBbMHtXKXUMUxH7zV3Zuozh+kUQWhHvg==", + "license": "MIT", + "optionalDependencies": { + "msgpackr-extract": "^3.0.2" + } + }, + "node_modules/msgpackr-extract": { + "version": "3.0.3", + "resolved": "https://registry.npmjs.org/msgpackr-extract/-/msgpackr-extract-3.0.3.tgz", + "integrity": "sha512-P0efT1C9jIdVRefqjzOQ9Xml57zpOXnIuS+csaB4MdZbTdmGDLo8XhzBG1N7aO11gKDDkJvBLULeFTo46wwreA==", + "hasInstallScript": true, + "license": "MIT", + "optional": true, + "dependencies": { + "node-gyp-build-optional-packages": "5.2.2" + }, + "bin": { + "download-msgpackr-prebuilds": "bin/download-prebuilds.js" + }, + "optionalDependencies": { + "@msgpackr-extract/msgpackr-extract-darwin-arm64": "3.0.3", + "@msgpackr-extract/msgpackr-extract-darwin-x64": "3.0.3", + "@msgpackr-extract/msgpackr-extract-linux-arm": "3.0.3", + "@msgpackr-extract/msgpackr-extract-linux-arm64": "3.0.3", + "@msgpackr-extract/msgpackr-extract-linux-x64": "3.0.3", + "@msgpackr-extract/msgpackr-extract-win32-x64": "3.0.3" + } + }, + "node_modules/multipasta": { + "version": "0.2.7", + "resolved": "https://registry.npmjs.org/multipasta/-/multipasta-0.2.7.tgz", + "integrity": "sha512-KPA58d68KgGil15oDqXjkUBEBYc00XvbPj5/X+dyzeo/lWm9Nc25pQRlf1D+gv4OpK7NM0J1odrbu9JNNGvynA==", + "license": "MIT" + }, + "node_modules/node-gyp-build-optional-packages": { + "version": "5.2.2", + "resolved": "https://registry.npmjs.org/node-gyp-build-optional-packages/-/node-gyp-build-optional-packages-5.2.2.tgz", + "integrity": "sha512-s+w+rBWnpTMwSFbaE0UXsRlg7hU4FjekKU4eyAih5T8nJuNZT1nNsskXpxmeqSK9UzkBl6UgRlnKc8hz8IEqOw==", + "license": "MIT", + "optional": true, + "dependencies": { + "detect-libc": "^2.0.1" + }, + "bin": { + "node-gyp-build-optional-packages": "bin.js", + "node-gyp-build-optional-packages-optional": "optional.js", + "node-gyp-build-optional-packages-test": "build-test.js" + } + }, "node_modules/path-key": { "version": "3.1.1", "resolved": "https://registry.npmjs.org/path-key/-/path-key-3.1.1.tgz", @@ -68,6 +276,22 @@ "node": ">=8" } }, + "node_modules/pure-rand": { + "version": "8.4.0", + "resolved": "https://registry.npmjs.org/pure-rand/-/pure-rand-8.4.0.tgz", + "integrity": "sha512-IoM8YF/jY0hiugFo/wOWqfmarlE6J0wc6fDK1PhftMk7MGhVZl88sZimmqBBFomLOCSmcCCpsfj7wXASCpvK9A==", + "funding": [ + { + "type": "individual", + "url": "https://github.com/sponsors/dubzzz" + }, + { + "type": "opencollective", + "url": "https://opencollective.com/fast-check" + } + ], + "license": "MIT" + }, "node_modules/shebang-command": { "version": "2.0.0", "resolved": "https://registry.npmjs.org/shebang-command/-/shebang-command-2.0.0.tgz", @@ -89,6 +313,28 @@ "node": ">=8" } }, + "node_modules/toml": { + "version": "4.1.1", + "resolved": "https://registry.npmjs.org/toml/-/toml-4.1.1.tgz", + "integrity": "sha512-EBJnVBr3dTXdA89WVFoAIPUqkBjxPMwRqsfuo1r240tKFHXv3zgca4+NJib/h6TyvGF7vOawz0jGuryJCdNHrw==", + "license": "MIT", + "engines": { + "node": ">=20" + } + }, + "node_modules/uuid": { + "version": "13.0.1", + "resolved": "https://registry.npmjs.org/uuid/-/uuid-13.0.1.tgz", + "integrity": "sha512-9ezox2roIft6ExBVTVqibSd5dc5/47Sw/uY6b4SjQUT2TzQ0tltNquWA46y4xPQmdZYqvnio22SgWd41M86+jw==", + "funding": [ + "https://github.com/sponsors/broofa", + "https://github.com/sponsors/ctavan" + ], + "license": "MIT", + "bin": { + "uuid": "dist-node/bin/uuid" + } + }, "node_modules/which": { "version": "2.0.2", "resolved": "https://registry.npmjs.org/which/-/which-2.0.2.tgz", @@ -104,6 +350,21 @@ "node": ">= 8" } }, + "node_modules/yaml": { + "version": "2.8.4", + "resolved": "https://registry.npmjs.org/yaml/-/yaml-2.8.4.tgz", + "integrity": "sha512-ml/JPOj9fOQK8RNnWojA67GbZ0ApXAUlN2UQclwv2eVgTgn7O9gg9o7paZWKMp4g0H3nTLtS9LVzhkpOFIKzog==", + "license": "ISC", + "bin": { + "yaml": "bin.mjs" + }, + "engines": { + "node": ">= 14.6" + }, + "funding": { + "url": "https://github.com/sponsors/eemeli" + } + }, "node_modules/zod": { "version": "4.1.8", "license": "MIT", diff --git a/cmd/server/app.go b/cmd/server/app.go index 4b24ec9..35553c1 100644 --- a/cmd/server/app.go +++ b/cmd/server/app.go @@ -8,6 +8,7 @@ import ( "with_you/internal/config" "with_you/internal/grpc/runner" + "with_you/internal/pkg/ws" "with_you/internal/router" "with_you/internal/service" @@ -24,6 +25,7 @@ type App struct { HotRankWorker *service.HotRankWorker GRPCServer *runner.Server Server *http.Server + WsHub *ws.Hub Logger *zap.Logger } @@ -35,6 +37,7 @@ func NewApp( pushService service.PushService, hotRankWorker *service.HotRankWorker, grpcServer *runner.Server, + wsHub *ws.Hub, logger *zap.Logger, ) *App { return &App{ @@ -44,6 +47,7 @@ func NewApp( PushService: pushService, HotRankWorker: hotRankWorker, GRPCServer: grpcServer, + WsHub: wsHub, Logger: logger, } } @@ -97,7 +101,13 @@ func (a *App) Start(ctx context.Context) error { func (a *App) Stop(ctx context.Context) error { a.Logger.Info("Shutting down server") - // 1. 停止 HTTP 服务器 + // 1. 关闭所有 WebSocket 连接 + if a.WsHub != nil { + a.Logger.Info("Closing WebSocket connections") + a.WsHub.CloseAllConnections() + } + + // 3. 停止 HTTP 服务器 if a.Server != nil { a.Logger.Info("Stopping HTTP server") if err := a.Server.Shutdown(ctx); err != nil { @@ -107,13 +117,13 @@ func (a *App) Stop(ctx context.Context) error { } } - // 2. 停止 gRPC 服务器 + // 4. 停止 gRPC 服务器 if a.GRPCServer != nil { a.Logger.Info("Stopping gRPC server") a.GRPCServer.Stop() } - // 3. 停止推送 Worker + // 5. 停止推送 Worker a.Logger.Info("Stopping push worker") a.PushService.StopPushWorker() @@ -122,7 +132,7 @@ func (a *App) Stop(ctx context.Context) error { a.HotRankWorker.Stop() } - // 4. 关闭数据库连接 + // 6. 关闭数据库连接 a.Logger.Info("Closing database connection") sqlDB, err := a.DB.DB() if err == nil { diff --git a/cmd/server/wire.go b/cmd/server/wire.go index a236d68..1e70fa7 100644 --- a/cmd/server/wire.go +++ b/cmd/server/wire.go @@ -55,47 +55,47 @@ func ProvideRouter( verificationHandler *handler.VerificationHandler, adminVerificationHandler *handler.AdminVerificationHandler, adminProfileAuditHandler *handler.AdminProfileAuditHandler, - setupHandler *handler.SetupHandler, + setupHandler *handler.SetupHandler, logService *service.LogService, activityService service.UserActivityService, casbinService service.CasbinService, wsHandler *handler.WSHandler, ) *router.Router { - return router.New( - userRepo, - userHandler, - postHandler, - commentHandler, - messageHandler, - notificationHandler, - uploadHandler, - jwtService, - pushHandler, - systemMessageHandler, - groupHandler, - stickerHandler, - voteHandler, - channelHandler, - scheduleHandler, - roleHandler, - adminUserHandler, - adminPostHandler, - adminCommentHandler, - adminGroupHandler, - adminDashboardHandler, - adminLogHandler, - qrcodeHandler, - materialHandler, - callHandler, - reportHandler, - adminReportHandler, - verificationHandler, - adminVerificationHandler, - adminProfileAuditHandler, - setupHandler, - logService, - activityService, - casbinService, - wsHandler, - ) + return router.New(router.RouterDeps{ + UserRepo: userRepo, + UserHandler: userHandler, + PostHandler: postHandler, + CommentHandler: commentHandler, + MessageHandler: messageHandler, + NotificationHandler: notificationHandler, + UploadHandler: uploadHandler, + JWTService: jwtService, + PushHandler: pushHandler, + SystemMessageHandler: systemMessageHandler, + GroupHandler: groupHandler, + StickerHandler: stickerHandler, + VoteHandler: voteHandler, + ChannelHandler: channelHandler, + ScheduleHandler: scheduleHandler, + RoleHandler: roleHandler, + AdminUserHandler: adminUserHandler, + AdminPostHandler: adminPostHandler, + AdminCommentHandler: adminCommentHandler, + AdminGroupHandler: adminGroupHandler, + AdminDashboardHandler: adminDashboardHandler, + AdminLogHandler: adminLogHandler, + QRCodeHandler: qrcodeHandler, + MaterialHandler: materialHandler, + CallHandler: callHandler, + ReportHandler: reportHandler, + AdminReportHandler: adminReportHandler, + VerificationHandler: verificationHandler, + AdminVerificationHandler: adminVerificationHandler, + AdminProfileAuditHandler: adminProfileAuditHandler, + SetupHandler: setupHandler, + LogService: logService, + ActivityService: activityService, + CasbinService: casbinService, + WSHandler: wsHandler, + }) } diff --git a/cmd/server/wire_gen.go b/cmd/server/wire_gen.go index 160703b..9708f14 100644 --- a/cmd/server/wire_gen.go +++ b/cmd/server/wire_gen.go @@ -149,7 +149,7 @@ func InitializeApp() (*App, error) { tradeHandler := handler.NewTradeHandler(tradeService) router := ProvideRouter(userRepository, userHandler, postHandler, commentHandler, messageHandler, notificationHandler, uploadHandler, jwtService, pushHandler, systemMessageHandler, groupHandler, stickerHandler, voteHandler, channelHandler, scheduleHandler, roleHandler, adminUserHandler, adminPostHandler, adminCommentHandler, adminGroupHandler, adminDashboardHandler, adminLogHandler, qrCodeHandler, materialHandler, callHandler, reportHandler, adminReportHandler, verificationHandler, adminVerificationHandler, adminProfileAuditHandler, setupHandler, tradeHandler, logService, userActivityService, casbinService, wsHandler) hotRankWorker := wire.ProvideHotRankWorker(config, postRepository, channelRepository, cache) - app := NewApp(config, db, router, pushService, hotRankWorker, server, logger) + app := NewApp(config, db, router, pushService, hotRankWorker, server, hub, logger) return app, nil } @@ -194,42 +194,42 @@ func ProvideRouter( casbinService service.CasbinService, wsHandler *handler.WSHandler, ) *router.Router { - return router.New( - userRepo, - userHandler, - postHandler, - commentHandler, - messageHandler, - notificationHandler, - uploadHandler, - jwtService, - pushHandler, - systemMessageHandler, - groupHandler, - stickerHandler, - voteHandler, - channelHandler, - scheduleHandler, - roleHandler, - adminUserHandler, - adminPostHandler, - adminCommentHandler, - adminGroupHandler, - adminDashboardHandler, - adminLogHandler, - qrcodeHandler, - materialHandler, - callHandler, - reportHandler, - adminReportHandler, - verificationHandler, - adminVerificationHandler, - adminProfileAuditHandler, - setupHandler, - tradeHandler, - logService, - activityService, - casbinService, - wsHandler, - ) + return router.New(router.RouterDeps{ + UserRepo: userRepo, + UserHandler: userHandler, + PostHandler: postHandler, + CommentHandler: commentHandler, + MessageHandler: messageHandler, + NotificationHandler: notificationHandler, + UploadHandler: uploadHandler, + JWTService: jwtService, + PushHandler: pushHandler, + SystemMessageHandler: systemMessageHandler, + GroupHandler: groupHandler, + StickerHandler: stickerHandler, + VoteHandler: voteHandler, + ChannelHandler: channelHandler, + ScheduleHandler: scheduleHandler, + RoleHandler: roleHandler, + AdminUserHandler: adminUserHandler, + AdminPostHandler: adminPostHandler, + AdminCommentHandler: adminCommentHandler, + AdminGroupHandler: adminGroupHandler, + AdminDashboardHandler: adminDashboardHandler, + AdminLogHandler: adminLogHandler, + QRCodeHandler: qrcodeHandler, + MaterialHandler: materialHandler, + CallHandler: callHandler, + ReportHandler: reportHandler, + AdminReportHandler: adminReportHandler, + VerificationHandler: verificationHandler, + AdminVerificationHandler: adminVerificationHandler, + AdminProfileAuditHandler: adminProfileAuditHandler, + SetupHandler: setupHandler, + TradeHandler: tradeHandler, + LogService: logService, + ActivityService: activityService, + CasbinService: casbinService, + WSHandler: wsHandler, + }) } diff --git a/internal/cache/cache.go b/internal/cache/cache.go index 802356c..2f2f892 100644 --- a/internal/cache/cache.go +++ b/internal/cache/cache.go @@ -32,6 +32,8 @@ type Cache interface { Increment(key string) int64 // IncrementBy 增加指定值 IncrementBy(key string, value int64) int64 + // DeleteBatch 批量删除多个 key（使用 Pipeline 减少网络往返） + DeleteBatch(keys []string) // ==================== Hash 操作 ==================== // HSet 设置 Hash 字段 diff --git a/internal/cache/conversation_cache.go b/internal/cache/conversation_cache.go index cd02dea..ceabeea 100644 --- a/internal/cache/conversation_cache.go +++ b/internal/cache/conversation_cache.go @@ -6,6 +6,8 @@ import ( "fmt" "time" + "github.com/redis/go-redis/v9" + "with_you/internal/model" "go.uber.org/zap" @@ -618,11 +620,11 @@ func (c *ConversationCache) GetMessagesBeforeSeq(ctx context.Context, convID str return messages, nil } -// CacheMessage 缓存单条消息（立即写入缓存） -// 写入 Hash、Sorted Set、更新计数 +// CacheMessage 缓存单条消息（使用 Redis Pipeline 减少网络往返） func (c *ConversationCache) CacheMessage(ctx context.Context, convID string, msg *model.Message) error { hashKey := MessageHashKey(convID) indexKey := MessageIndexKey(convID) + countKey := MessageCountKey(convID) msgData := MessageCacheDataFromModel(msg) data, err := json.Marshal(msgData) @@ -630,23 +632,34 @@ func (c *ConversationCache) CacheMessage(ctx context.Context, convID string, msg return fmt.Errorf("failed to marshal message: %w", err) } - // HSET 消息详情 - if err := c.cache.HSet(ctx, hashKey, fmt.Sprintf("%d", msg.Seq), string(data)); err != nil { + hashKey = ResolveKey(hashKey) + indexKey = ResolveKey(indexKey) + countKey = ResolveKey(countKey) + + redisCache, ok := c.cache.(*RedisCache) + if ok { + rdb := redisCache.GetRedisClient().GetClient() + pipe := rdb.Pipeline() + pipe.HSet(ctx, hashKey, fmt.Sprintf("%d", msg.Seq), string(data)) + pipe.ZAdd(ctx, indexKey, redis.Z{Score: float64(msg.Seq), Member: fmt.Sprintf("%d", msg.Seq)}) + pipe.Expire(ctx, hashKey, c.settings.MessageDetailTTL) + pipe.Expire(ctx, indexKey, c.settings.MessageIndexTTL) + pipe.Incr(ctx, countKey) + if _, err := pipe.Exec(ctx); err != nil { + return fmt.Errorf("failed to pipeline cache message: %w", err) + } + return nil + } + + if err := c.cache.HSet(ctx, MessageHashKey(convID), fmt.Sprintf("%d", msg.Seq), string(data)); err != nil { return fmt.Errorf("failed to set hash: %w", err) } - - // ZADD 消息索引 - if err := c.cache.ZAdd(ctx, indexKey, float64(msg.Seq), fmt.Sprintf("%d", msg.Seq)); err != nil { + if err := c.cache.ZAdd(ctx, MessageIndexKey(convID), float64(msg.Seq), fmt.Sprintf("%d", msg.Seq)); err != nil { return fmt.Errorf("failed to add to sorted set: %w", err) } - - // 设置 TTL - c.cache.Expire(ctx, hashKey, c.settings.MessageDetailTTL) - c.cache.Expire(ctx, indexKey, c.settings.MessageIndexTTL) - - // INCR 消息计数 + c.cache.Expire(ctx, MessageHashKey(convID), c.settings.MessageDetailTTL) + c.cache.Expire(ctx, MessageIndexKey(convID), c.settings.MessageIndexTTL) c.cache.Incr(ctx, MessageCountKey(convID)) - return nil } diff --git a/internal/cache/keys.go b/internal/cache/keys.go index e9a5a2f..6ae5598 100644 --- a/internal/cache/keys.go +++ b/internal/cache/keys.go @@ -35,11 +35,12 @@ const ( PrefixChannelsAllList = "channels:all_list" // 消息缓存相关 - keyPrefixMsgHash = "msg_hash" // 消息详情 Hash - keyPrefixMsgIndex = "msg_index" // 消息索引 Sorted Set - keyPrefixMsgCount = "msg_count" // 消息计数 - keyPrefixMsgSeq = "msg_seq" // Seq 计数器 - keyPrefixMsgPage = "msg_page" // 分页缓存 + keyPrefixMsgHash = "msg_hash" // 消息详情 Hash + keyPrefixMsgIndex = "msg_index" // 消息索引 Sorted Set + keyPrefixMsgCount = "msg_count" // 消息计数 + keyPrefixMsgSeq = "msg_seq" // Seq 计数器 + keyPrefixMsgPage = "msg_page" // 分页缓存 + keyPrefixMsgIdempotent = "msg_idem" // 消息幂等键 ) // PostListKey 生成帖子列表缓存键 @@ -221,3 +222,13 @@ func MessagePageKey(convID string, page, pageSize int) string { func InvalidateMessagePages(cache Cache, conversationID string) { cache.DeleteByPrefix(fmt.Sprintf("%s:%s:", keyPrefixMsgPage, conversationID)) } + +// MessageIdempotentKey 消息幂等键 (clientMsgID -> server MsgID) +func MessageIdempotentKey(senderID, clientMsgID string) string { + return fmt.Sprintf("%s:%s:%s", keyPrefixMsgIdempotent, senderID, clientMsgID) +} + +// PushDedupKey 推送去重键，防止同一消息重复推送给同一用户 +func PushDedupKey(userID, messageID string) string { + return fmt.Sprintf("push_dedup:%s:%s", userID, messageID) +} diff --git a/internal/cache/layered_cache.go b/internal/cache/layered_cache.go index b63240c..4455727 100644 --- a/internal/cache/layered_cache.go +++ b/internal/cache/layered_cache.go @@ -111,7 +111,11 @@ func (c *LayeredCache) Get(key string) (any, bool) { c.mu.Unlock() if ttl := c.getLocalTTL(); ttl > 0 { - c.local.Set(key, val, ttl) + if str, ok := val.(string); ok { + c.local.SetRaw(key, []byte(str), ttl) + } else { + c.local.Set(key, val, ttl) + } } return val, true } @@ -157,6 +161,23 @@ func (c *LayeredCache) DeleteByPrefix(prefix string) { } } +func (c *LayeredCache) DeleteBatch(keys []string) { + if !c.enabled || len(keys) == 0 { + return + } + + c.mu.Lock() + c.stats.Invalidates += int64(len(keys)) + c.mu.Unlock() + + for _, k := range keys { + c.local.Delete(k) + } + if c.redis != nil { + c.redis.DeleteBatch(keys) + } +} + func (c *LayeredCache) Clear() { c.local.Clear() if c.redis != nil { diff --git a/internal/cache/lru.go b/internal/cache/lru.go index 8131eb6..3e7e3d5 100644 --- a/internal/cache/lru.go +++ b/internal/cache/lru.go @@ -87,6 +87,39 @@ func (l *LRU) set(key string, value any, ttl time.Duration) { } } +// SetRaw stores raw bytes directly without re-serializing. +// This avoids double-encoding when promoting values from Redis to local cache. +func (l *LRU) SetRaw(key string, raw []byte, ttl time.Duration) { + var expires time.Time + if ttl > 0 { + expires = time.Now().Add(ttl) + } + + l.lock.Lock() + defer l.lock.Unlock() + + if ent, ok := l.items[key]; ok { + l.evictList.MoveToFront(ent) + e := ent.Value.(*lruEntry) + e.value = raw + e.expires = expires + if e.generation != l.currentGeneration { + e.generation = l.currentGeneration + l.len++ + } + return + } + + ent := &lruEntry{key, raw, expires, l.currentGeneration} + entry := l.evictList.PushFront(ent) + l.items[key] = entry + l.len++ + + if l.evictList.Len() > l.size { + l.removeElement(l.evictList.Back()) + } +} + func (l *LRU) Get(key string) (any, bool) { val, err := l.getItem(key) if err != nil { diff --git a/internal/cache/lru_striped.go b/internal/cache/lru_striped.go index c921c85..c76ad2d 100644 --- a/internal/cache/lru_striped.go +++ b/internal/cache/lru_striped.go @@ -106,11 +106,20 @@ func (l *LRUStriped) Name() string { } func (l *LRUStriped) DeleteMulti(keys []string) error { - var err error for _, key := range keys { l.Delete(key) } - return err + return nil +} + +func (l *LRUStriped) DeleteBatch(keys []string) { + for _, key := range keys { + l.Delete(key) + } +} + +func (l *LRUStriped) SetRaw(key string, raw []byte, ttl time.Duration) { + l.keyBucket(key).SetRaw(key, raw, ttl) } func (l *LRUStriped) GetMulti(keys []string, values []any) []error { diff --git a/internal/cache/redis_cache.go b/internal/cache/redis_cache.go index 353d789..05814ec 100644 --- a/internal/cache/redis_cache.go +++ b/internal/cache/redis_cache.go @@ -82,7 +82,6 @@ func (c *RedisCache) Delete(key string) { // DeleteByPrefix 根据前缀删除缓存 func (c *RedisCache) DeleteByPrefix(prefix string) { prefix = normalizePrefix(prefix) - // 使用原生客户端执行SCAN命令 rdb := c.client.GetClient() var cursor uint64 for { @@ -112,6 +111,30 @@ func (c *RedisCache) DeleteByPrefix(prefix string) { } } +// DeleteBatch 批量删除多个 key（使用 Redis Pipeline 减少网络往返） +func (c *RedisCache) DeleteBatch(keys []string) { + if len(keys) == 0 { + return + } + normalizedKeys := make([]string, len(keys)) + for i, k := range keys { + normalizedKeys[i] = normalizeKey(k) + } + + recordInvalidateMultiple(int64(len(normalizedKeys))) + // 使用 pipeline 批量删除 + pipe := c.client.GetClient().Pipeline() + for _, k := range normalizedKeys { + pipe.Del(c.ctx, k) + } + if _, err := pipe.Exec(c.ctx); err != nil { + zap.L().Error("Failed to batch delete keys", + zap.Int("count", len(normalizedKeys)), + zap.Error(err), + ) + } +} + // Clear 清空所有缓存 func (c *RedisCache) Clear() { if settings.DisableFlushDB { diff --git a/internal/config/config.go b/internal/config/config.go index 846816d..283a4fe 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -1,10 +1,7 @@ package config import ( - "cmp" "fmt" - "os" - "strconv" "strings" "time" @@ -149,14 +146,6 @@ func Load(configPath string) (*Config, error) { viper.SetDefault("grpc.tls_enabled", false) viper.SetDefault("grpc.tls_cert_file", "") viper.SetDefault("grpc.tls_key_file", "") - // Casbin 默认值 - viper.SetDefault("casbin.model_path", "./configs/casbin/model.conf") - viper.SetDefault("casbin.auto_save", true) - viper.SetDefault("casbin.auto_load", true) - viper.SetDefault("casbin.enable_cache", true) - viper.SetDefault("casbin.cache_ttl", 300) - viper.SetDefault("casbin.skip_routes", []string{"/health", "/api/v1/auth/login", "/api/v1/auth/register"}) - // Casbin 默认值 viper.SetDefault("casbin.model_path", "./configs/casbin/model.conf") viper.SetDefault("casbin.auto_save", true) viper.SetDefault("casbin.auto_load", true) @@ -198,96 +187,7 @@ func Load(configPath string) (*Config, error) { cfg.JWT.AccessTokenExpire = time.Duration(viper.GetInt("jwt.access_token_expire")) * time.Second cfg.JWT.RefreshTokenExpire = time.Duration(viper.GetInt("jwt.refresh_token_expire")) * time.Second - // 环境变量覆盖（显式处理敏感配置） - cfg.JWT.Secret = getEnvOrDefault("APP_JWT_SECRET", cfg.JWT.Secret) - cfg.Database.SQLite.Path = getEnvOrDefault("APP_DATABASE_SQLITE_PATH", cfg.Database.SQLite.Path) - cfg.Database.Postgres.Host = getEnvOrDefault("APP_DATABASE_POSTGRES_HOST", cfg.Database.Postgres.Host) - cfg.Database.Postgres.Port, _ = strconv.Atoi(getEnvOrDefault("APP_DATABASE_POSTGRES_PORT", fmt.Sprintf("%d", cfg.Database.Postgres.Port))) - cfg.Database.Postgres.User = getEnvOrDefault("APP_DATABASE_POSTGRES_USER", cfg.Database.Postgres.User) - cfg.Database.Postgres.Password = getEnvOrDefault("APP_DATABASE_POSTGRES_PASSWORD", cfg.Database.Postgres.Password) - cfg.Database.Postgres.DBName = getEnvOrDefault("APP_DATABASE_POSTGRES_DBNAME", cfg.Database.Postgres.DBName) - cfg.Database.LogLevel = getEnvOrDefault("APP_DATABASE_LOG_LEVEL", cfg.Database.LogLevel) - cfg.Database.SlowThresholdMs, _ = strconv.Atoi(getEnvOrDefault("APP_DATABASE_SLOW_THRESHOLD_MS", fmt.Sprintf("%d", cfg.Database.SlowThresholdMs))) - cfg.Database.IgnoreRecordNotFound, _ = strconv.ParseBool(getEnvOrDefault("APP_DATABASE_IGNORE_RECORD_NOT_FOUND", fmt.Sprintf("%t", cfg.Database.IgnoreRecordNotFound))) - cfg.Database.ParameterizedQueries, _ = strconv.ParseBool(getEnvOrDefault("APP_DATABASE_PARAMETERIZED_QUERIES", fmt.Sprintf("%t", cfg.Database.ParameterizedQueries))) - cfg.Redis.Redis.Host = getEnvOrDefault("APP_REDIS_REDIS_HOST", cfg.Redis.Redis.Host) - cfg.Redis.Redis.Port, _ = strconv.Atoi(getEnvOrDefault("APP_REDIS_REDIS_PORT", fmt.Sprintf("%d", cfg.Redis.Redis.Port))) - cfg.Redis.Redis.Password = getEnvOrDefault("APP_REDIS_REDIS_PASSWORD", cfg.Redis.Redis.Password) - cfg.Redis.Redis.DB, _ = strconv.Atoi(getEnvOrDefault("APP_REDIS_REDIS_DB", fmt.Sprintf("%d", cfg.Redis.Redis.DB))) - cfg.Redis.Miniredis.Host = getEnvOrDefault("APP_REDIS_MINIREDIS_HOST", cfg.Redis.Miniredis.Host) - cfg.Redis.Miniredis.Port, _ = strconv.Atoi(getEnvOrDefault("APP_REDIS_MINIREDIS_PORT", fmt.Sprintf("%d", cfg.Redis.Miniredis.Port))) - cfg.Redis.Type = getEnvOrDefault("APP_REDIS_TYPE", cfg.Redis.Type) - cfg.Cache.KeyPrefix = getEnvOrDefault("APP_CACHE_KEY_PREFIX", cfg.Cache.KeyPrefix) - cfg.Cache.Enabled, _ = strconv.ParseBool(getEnvOrDefault("APP_CACHE_ENABLED", fmt.Sprintf("%t", cfg.Cache.Enabled))) - cfg.Cache.DisableFlushDB, _ = strconv.ParseBool(getEnvOrDefault("APP_CACHE_DISABLE_FLUSHDB", fmt.Sprintf("%t", cfg.Cache.DisableFlushDB))) - cfg.Cache.DefaultTTL, _ = strconv.Atoi(getEnvOrDefault("APP_CACHE_DEFAULT_TTL", fmt.Sprintf("%d", cfg.Cache.DefaultTTL))) - cfg.Cache.NullTTL, _ = strconv.Atoi(getEnvOrDefault("APP_CACHE_NULL_TTL", fmt.Sprintf("%d", cfg.Cache.NullTTL))) - cfg.Cache.JitterRatio, _ = strconv.ParseFloat(getEnvOrDefault("APP_CACHE_JITTER_RATIO", fmt.Sprintf("%.2f", cfg.Cache.JitterRatio)), 64) - cfg.Cache.Modules.PostList, _ = strconv.Atoi(getEnvOrDefault("APP_CACHE_MODULES_POST_LIST_TTL", fmt.Sprintf("%d", cfg.Cache.Modules.PostList))) - cfg.Cache.Modules.Conversation, _ = strconv.Atoi(getEnvOrDefault("APP_CACHE_MODULES_CONVERSATION_TTL", fmt.Sprintf("%d", cfg.Cache.Modules.Conversation))) - cfg.Cache.Modules.UnreadCount, _ = strconv.Atoi(getEnvOrDefault("APP_CACHE_MODULES_UNREAD_COUNT_TTL", fmt.Sprintf("%d", cfg.Cache.Modules.UnreadCount))) - cfg.Cache.Modules.GroupMembers, _ = strconv.Atoi(getEnvOrDefault("APP_CACHE_MODULES_GROUP_MEMBERS_TTL", fmt.Sprintf("%d", cfg.Cache.Modules.GroupMembers))) - cfg.S3.Endpoint = getEnvOrDefault("APP_S3_ENDPOINT", cfg.S3.Endpoint) - cfg.S3.AccessKey = getEnvOrDefault("APP_S3_ACCESS_KEY", cfg.S3.AccessKey) - cfg.S3.SecretKey = getEnvOrDefault("APP_S3_SECRET_KEY", cfg.S3.SecretKey) - cfg.S3.Bucket = getEnvOrDefault("APP_S3_BUCKET", cfg.S3.Bucket) - cfg.S3.Domain = getEnvOrDefault("APP_S3_DOMAIN", cfg.S3.Domain) - cfg.Server.Host = getEnvOrDefault("APP_SERVER_HOST", cfg.Server.Host) - cfg.Server.Port, _ = strconv.Atoi(getEnvOrDefault("APP_SERVER_PORT", fmt.Sprintf("%d", cfg.Server.Port))) - cfg.Server.Mode = getEnvOrDefault("APP_SERVER_MODE", cfg.Server.Mode) - cfg.OpenAI.BaseURL = getEnvOrDefault("APP_OPENAI_BASE_URL", cfg.OpenAI.BaseURL) - cfg.OpenAI.APIKey = getEnvOrDefault("APP_OPENAI_API_KEY", cfg.OpenAI.APIKey) - cfg.OpenAI.ModerationModel = getEnvOrDefault("APP_OPENAI_MODERATION_MODEL", cfg.OpenAI.ModerationModel) - cfg.OpenAI.ModerationMaxImagesPerRequest, _ = strconv.Atoi(getEnvOrDefault("APP_OPENAI_MODERATION_MAX_IMAGES_PER_REQUEST", fmt.Sprintf("%d", cfg.OpenAI.ModerationMaxImagesPerRequest))) - cfg.OpenAI.RequestTimeout, _ = strconv.Atoi(getEnvOrDefault("APP_OPENAI_REQUEST_TIMEOUT", fmt.Sprintf("%d", cfg.OpenAI.RequestTimeout))) - cfg.OpenAI.Enabled, _ = strconv.ParseBool(getEnvOrDefault("APP_OPENAI_ENABLED", fmt.Sprintf("%t", cfg.OpenAI.Enabled))) - cfg.OpenAI.StrictModeration, _ = strconv.ParseBool(getEnvOrDefault("APP_OPENAI_STRICT_MODERATION", fmt.Sprintf("%t", cfg.OpenAI.StrictModeration))) - // TencentTMS 环境变量覆盖 - cfg.TencentTMS.Enabled, _ = strconv.ParseBool(getEnvOrDefault("APP_TENCENT_TMS_ENABLED", fmt.Sprintf("%t", cfg.TencentTMS.Enabled))) - cfg.TencentTMS.SecretID = getEnvOrDefault("APP_TENCENT_TMS_SECRET_ID", cfg.TencentTMS.SecretID) - cfg.TencentTMS.SecretKey = getEnvOrDefault("APP_TENCENT_TMS_SECRET_KEY", cfg.TencentTMS.SecretKey) - cfg.TencentTMS.Region = getEnvOrDefault("APP_TENCENT_TMS_REGION", cfg.TencentTMS.Region) - cfg.TencentTMS.BizType = getEnvOrDefault("APP_TENCENT_TMS_BIZ_TYPE", cfg.TencentTMS.BizType) - cfg.TencentTMS.Timeout, _ = strconv.Atoi(getEnvOrDefault("APP_TENCENT_TMS_TIMEOUT", fmt.Sprintf("%d", cfg.TencentTMS.Timeout))) - cfg.Email.Enabled, _ = strconv.ParseBool(getEnvOrDefault("APP_EMAIL_ENABLED", fmt.Sprintf("%t", cfg.Email.Enabled))) - cfg.Email.Host = getEnvOrDefault("APP_EMAIL_HOST", cfg.Email.Host) - cfg.Email.Port, _ = strconv.Atoi(getEnvOrDefault("APP_EMAIL_PORT", fmt.Sprintf("%d", cfg.Email.Port))) - cfg.Email.Username = getEnvOrDefault("APP_EMAIL_USERNAME", cfg.Email.Username) - cfg.Email.Password = getEnvOrDefault("APP_EMAIL_PASSWORD", cfg.Email.Password) - cfg.Email.FromAddress = getEnvOrDefault("APP_EMAIL_FROM_ADDRESS", cfg.Email.FromAddress) - cfg.Email.FromName = getEnvOrDefault("APP_EMAIL_FROM_NAME", cfg.Email.FromName) - cfg.Email.UseTLS, _ = strconv.ParseBool(getEnvOrDefault("APP_EMAIL_USE_TLS", fmt.Sprintf("%t", cfg.Email.UseTLS))) - cfg.Email.InsecureSkipVerify, _ = strconv.ParseBool(getEnvOrDefault("APP_EMAIL_INSECURE_SKIP_VERIFY", fmt.Sprintf("%t", cfg.Email.InsecureSkipVerify))) - cfg.Email.Timeout, _ = strconv.Atoi(getEnvOrDefault("APP_EMAIL_TIMEOUT", fmt.Sprintf("%d", cfg.Email.Timeout))) - // GRPC 环境变量覆盖 - cfg.GRPC.Enabled, _ = strconv.ParseBool(getEnvOrDefault("APP_GRPC_ENABLED", fmt.Sprintf("%t", cfg.GRPC.Enabled))) - cfg.GRPC.Port, _ = strconv.Atoi(getEnvOrDefault("APP_GRPC_PORT", fmt.Sprintf("%d", cfg.GRPC.Port))) - cfg.GRPC.TLSEnabled, _ = strconv.ParseBool(getEnvOrDefault("APP_GRPC_TLS_ENABLED", fmt.Sprintf("%t", cfg.GRPC.TLSEnabled))) - cfg.GRPC.TLSCertFile = getEnvOrDefault("APP_GRPC_TLS_CERT_FILE", cfg.GRPC.TLSCertFile) - cfg.GRPC.TLSKeyFile = getEnvOrDefault("APP_GRPC_TLS_KEY_FILE", cfg.GRPC.TLSKeyFile) - // Casbin 环境变量覆盖 - cfg.Casbin.ModelPath = getEnvOrDefault("APP_CASBIN_MODEL_PATH", cfg.Casbin.ModelPath) - cfg.Casbin.AutoSave, _ = strconv.ParseBool(getEnvOrDefault("APP_CASBIN_AUTO_SAVE", fmt.Sprintf("%t", cfg.Casbin.AutoSave))) - cfg.Casbin.AutoLoad, _ = strconv.ParseBool(getEnvOrDefault("APP_CASBIN_AUTO_LOAD", fmt.Sprintf("%t", cfg.Casbin.AutoLoad))) - cfg.Casbin.EnableCache, _ = strconv.ParseBool(getEnvOrDefault("APP_CASBIN_ENABLE_CACHE", fmt.Sprintf("%t", cfg.Casbin.EnableCache))) - cfg.Casbin.CacheTTL, _ = strconv.Atoi(getEnvOrDefault("APP_CASBIN_CACHE_TTL", fmt.Sprintf("%d", cfg.Casbin.CacheTTL))) - // Encryption 环境变量覆盖 - cfg.Encryption.Enabled, _ = strconv.ParseBool(getEnvOrDefault("APP_ENCRYPTION_ENABLED", fmt.Sprintf("%t", cfg.Encryption.Enabled))) - cfg.Encryption.Key = getEnvOrDefault("APP_ENCRYPTION_KEY", cfg.Encryption.Key) - cfg.Encryption.KeyVersion, _ = strconv.Atoi(getEnvOrDefault("APP_ENCRYPTION_KEY_VERSION", fmt.Sprintf("%d", cfg.Encryption.KeyVersion))) - // Sensitive 环境变量覆盖 - cfg.Sensitive.Enabled, _ = strconv.ParseBool(getEnvOrDefault("APP_SENSITIVE_ENABLED", fmt.Sprintf("%t", cfg.Sensitive.Enabled))) - cfg.Sensitive.ReplaceStr = getEnvOrDefault("APP_SENSITIVE_REPLACE_STR", cfg.Sensitive.ReplaceStr) - cfg.Sensitive.MinMatchLen, _ = strconv.Atoi(getEnvOrDefault("APP_SENSITIVE_MIN_MATCH_LEN", fmt.Sprintf("%d", cfg.Sensitive.MinMatchLen))) - cfg.Sensitive.LoadFromDB, _ = strconv.ParseBool(getEnvOrDefault("APP_SENSITIVE_LOAD_FROM_DB", fmt.Sprintf("%t", cfg.Sensitive.LoadFromDB))) - cfg.Sensitive.LoadFromRedis, _ = strconv.ParseBool(getEnvOrDefault("APP_SENSITIVE_LOAD_FROM_REDIS", fmt.Sprintf("%t", cfg.Sensitive.LoadFromRedis))) - cfg.Sensitive.RedisKeyPrefix = getEnvOrDefault("APP_SENSITIVE_REDIS_KEY_PREFIX", cfg.Sensitive.RedisKeyPrefix) - cfg.JPush.Enabled, _ = strconv.ParseBool(getEnvOrDefault("APP_JPUSH_ENABLED", fmt.Sprintf("%t", cfg.JPush.Enabled))) - cfg.JPush.AppKey = getEnvOrDefault("APP_JPUSH_APP_KEY", cfg.JPush.AppKey) - cfg.JPush.MasterSecret = getEnvOrDefault("APP_JPUSH_MASTER_SECRET", cfg.JPush.MasterSecret) - cfg.JPush.Production, _ = strconv.ParseBool(getEnvOrDefault("APP_JPUSH_PRODUCTION", fmt.Sprintf("%t", cfg.JPush.Production))) - - cfg.SetupSecret = getEnvOrDefault("APP_SETUP_SECRET", cfg.SetupSecret) + // Viper AutomaticEnv 已自动处理所有 APP_ 前缀的环境变量 // 安全检查：生产模式必须设置JWT密钥 if cfg.Server.Mode != "debug" { @@ -302,7 +202,4 @@ func Load(configPath string) (*Config, error) { return &cfg, nil } -// getEnvOrDefault 获取环境变量值，如果未设置则返回默认值 -func getEnvOrDefault(key, defaultValue string) string { - return cmp.Or(os.Getenv(key), defaultValue) -} + diff --git a/internal/dto/dto.go b/internal/dto/dto.go index fc4852f..55af1f9 100644 --- a/internal/dto/dto.go +++ b/internal/dto/dto.go @@ -177,33 +177,6 @@ type PostResponse struct { Channel *PostChannelBrief `json:"channel,omitempty"` } -// PostDetailResponse 帖子详情响应 -type PostDetailResponse struct { - ID string `json:"id"` - UserID string `json:"user_id"` - ChannelID *string `json:"channel_id,omitempty"` - Title string `json:"title"` - Content string `json:"content"` - Segments model.MessageSegments `json:"segments,omitempty"` - Images []PostImageResponse `json:"images"` - Status string `json:"status"` - LikesCount int `json:"likes_count"` - CommentsCount int `json:"comments_count"` - FavoritesCount int `json:"favorites_count"` - SharesCount int `json:"shares_count"` - ViewsCount int `json:"views_count"` - IsPinned bool `json:"is_pinned"` - IsLocked bool `json:"is_locked"` - IsVote bool `json:"is_vote"` - CreatedAt string `json:"created_at"` - UpdatedAt string `json:"updated_at"` - ContentEditedAt string `json:"content_edited_at,omitempty"` - Author *UserResponse `json:"author"` - IsLiked bool `json:"is_liked"` - IsFavorited bool `json:"is_favorited"` - Channel *PostChannelBrief `json:"channel,omitempty"` -} - // ==================== Comment DTOs ==================== // CommentImageResponse 评论图片响应 @@ -454,8 +427,9 @@ type CreateConversationRequest struct { // SendMessageRequest 发送消息请求 type SendMessageRequest struct { - Segments model.MessageSegments `json:"segments" binding:"required"` // 消息链（必须） - ReplyToID *string `json:"reply_to_id,omitempty"` // 回复的消息ID (string类型) + Segments model.MessageSegments `json:"segments" binding:"required"` // 消息链（必须） + ReplyToID *string `json:"reply_to_id,omitempty"` // 回复的消息ID (string类型) + ClientMsgID string `json:"client_msg_id,omitempty"` // 客户端消息ID（幂等性去重） } // MarkReadRequest 标记已读请求 @@ -751,10 +725,11 @@ type WSEventResponse struct { // SendMessageParams 发送消息参数（用于 REST API） type SendMessageParams struct { - DetailType string `json:"detail_type"` // 消息类型: private, group - ConversationID string `json:"conversation_id"` // 会话ID - Segments model.MessageSegments `json:"segments"` // 消息内容（消息段数组） - ReplyToID *string `json:"reply_to_id,omitempty"` // 回复的消息ID + DetailType string `json:"detail_type"` // 消息类型: private, group + ConversationID string `json:"conversation_id"` // 会话ID + Segments model.MessageSegments `json:"segments"` // 消息内容（消息段数组） + ReplyToID *string `json:"reply_to_id,omitempty"` // 回复的消息ID + ClientMsgID string `json:"client_msg_id,omitempty"` // 客户端消息ID（幂等性去重） } // DeleteMsgParams 撤回消息参数 diff --git a/internal/dto/post_converter.go b/internal/dto/post_converter.go index 4391c0d..4294b27 100644 --- a/internal/dto/post_converter.go +++ b/internal/dto/post_converter.go @@ -124,50 +124,7 @@ func ConvertPostToResponse(post *model.Post, channelByID map[string]*model.Chann } } -// ConvertPostToDetailResponse 将Post转换为PostDetailResponse；channelByID 可为 nil -func ConvertPostToDetailResponse(post *model.Post, channelByID map[string]*model.Channel, isLiked, isFavorited bool) *PostDetailResponse { - if post == nil { - return nil - } - - images := make([]PostImageResponse, 0) - for _, img := range post.Images { - images = append(images, ConvertPostImageToResponse(&img)) - } - - var author *UserResponse - if post.User != nil { - author = ConvertUserToResponse(post.User) - } - - return &PostDetailResponse{ - ID: post.ID, - UserID: post.UserID, - ChannelID: post.ChannelID, - Title: post.Title, - Content: post.Content, - Segments: SegmentsOrDefault(post.Segments, post.Content), - Images: images, - Status: string(post.Status), - LikesCount: post.LikesCount, - CommentsCount: post.CommentsCount, - FavoritesCount: post.FavoritesCount, - SharesCount: post.SharesCount, - ViewsCount: post.ViewsCount, - IsPinned: post.IsPinned, - IsLocked: post.IsLocked, - IsVote: post.IsVote, - CreatedAt: FormatTime(post.CreatedAt), - UpdatedAt: FormatTime(post.UpdatedAt), - ContentEditedAt: FormatTimePointer(post.ContentEditedAt), - Author: author, - IsLiked: isLiked, - IsFavorited: isFavorited, - Channel: postChannelBrief(post, channelByID), - } -} - -// ConvertPostsToResponse 将Post列表转换为响应列表（每个帖子独立检查点赞/收藏状态）；channelByID 可为 nil +// ConvertPostsToResponse 将Post列表转换为响应列表；channelByID 可为 nil func ConvertPostsToResponse(posts []*model.Post, channelByID map[string]*model.Channel, isLikedMap, isFavoritedMap map[string]bool) []*PostResponse { result := make([]*PostResponse, 0, len(posts)) for _, post := range posts { diff --git a/internal/dto/report.go b/internal/dto/report.go index cc9db95..aac5f82 100644 --- a/internal/dto/report.go +++ b/internal/dto/report.go @@ -108,6 +108,12 @@ type AdminBatchHandleResponse struct { // ==================== 转换函数 ==================== +func baseAdminReportFields(report *model.Report) (string, string) { + reason := string(report.Reason) + status := string(report.Status) + return reason, status +} + // ConvertReportToResponse 将 Report 模型转换为响应 func ConvertReportToResponse(report *model.Report) *ReportResponse { return &ReportResponse{ @@ -122,6 +128,19 @@ func ConvertReportToResponse(report *model.Report) *ReportResponse { } } +func fillAdminReportCommon(resp *AdminReportListResponse, report *model.Report, reporter *model.User, handler *model.User) { + if reporter != nil { + resp.Reporter = ConvertUserToResponse(reporter) + } + if handler != nil { + resp.Handler = ConvertUserToResponse(handler) + } + if report.HandledAt != nil { + handledAt := FormatTime(*report.HandledAt) + resp.HandledAt = &handledAt + } +} + // ConvertReportToAdminListResponse 将 Report 转换为管理端列表响应 func ConvertReportToAdminListResponse(report *model.Report, reporter *model.User, handler *model.User) *AdminReportListResponse { resp := &AdminReportListResponse{ @@ -136,51 +155,34 @@ func ConvertReportToAdminListResponse(report *model.Report, reporter *model.User Result: report.Result, CreatedAt: FormatTime(report.CreatedAt), } - - if reporter != nil { - resp.Reporter = ConvertUserToResponse(reporter) - } - - if handler != nil { - resp.Handler = ConvertUserToResponse(handler) - } - - if report.HandledAt != nil { - handledAt := FormatTime(*report.HandledAt) - resp.HandledAt = &handledAt - } - + fillAdminReportCommon(resp, report, reporter, handler) return resp } // ConvertReportToAdminDetailResponse 将 Report 转换为管理端详情响应 func ConvertReportToAdminDetailResponse(report *model.Report, reporter *model.User, handler *model.User, targetContent *ReportTargetContent) *AdminReportDetailResponse { resp := &AdminReportDetailResponse{ - ID: report.ID, - ReporterID: report.ReporterID, - TargetType: string(report.TargetType), - TargetID: report.TargetID, - Reason: string(report.Reason), - Description: report.Description, - Status: string(report.Status), - HandledBy: report.HandledBy, - Result: report.Result, - CreatedAt: FormatTime(report.CreatedAt), + ID: report.ID, + ReporterID: report.ReporterID, + TargetType: string(report.TargetType), + TargetID: report.TargetID, + Reason: string(report.Reason), + Description: report.Description, + Status: string(report.Status), + HandledBy: report.HandledBy, + Result: report.Result, + CreatedAt: FormatTime(report.CreatedAt), TargetContent: targetContent, } - if reporter != nil { resp.Reporter = ConvertUserToResponse(reporter) } - if handler != nil { resp.Handler = ConvertUserToResponse(handler) } - if report.HandledAt != nil { handledAt := FormatTime(*report.HandledAt) resp.HandledAt = &handledAt } - return resp } \ No newline at end of file diff --git a/internal/dto/trade_converter.go b/internal/dto/trade_converter.go index e673b46..6832dd7 100644 --- a/internal/dto/trade_converter.go +++ b/internal/dto/trade_converter.go @@ -45,8 +45,8 @@ func ConvertTradeItemToResponse(item *model.TradeItem, isFavorited bool) *TradeI FavoritesCount: item.FavoritesCount, IsFavorited: isFavorited, Images: ConvertTradeImagesToResponse(item.Images), - CreatedAt: item.CreatedAt.Format("2006-01-02T15:04:05Z07:00"), - UpdatedAt: item.UpdatedAt.Format("2006-01-02T15:04:05Z07:00"), + CreatedAt: FormatTime(item.CreatedAt), + UpdatedAt: FormatTime(item.UpdatedAt), } if len(item.Segments) > 0 { resp.Segments = item.Segments @@ -66,32 +66,6 @@ func ConvertTradeItemsToResponse(items []*model.TradeItem, favoriteMap map[strin return result } -func ConvertTradeItemToDetailResponse(item *model.TradeItem, isFavorited bool) *TradeItemDetailResponse { - if item == nil { - return nil - } - resp := &TradeItemDetailResponse{ - ID: item.ID, - UserID: item.UserID, - TradeType: string(item.TradeType), - Category: string(item.Category), - Title: item.Title, - Content: item.Content, - Price: item.Price, - Condition: item.Condition, - Status: string(item.Status), - ViewsCount: item.ViewsCount, - FavoritesCount: item.FavoritesCount, - IsFavorited: isFavorited, - Images: ConvertTradeImagesToResponse(item.Images), - CreatedAt: item.CreatedAt.Format("2006-01-02T15:04:05Z07:00"), - UpdatedAt: item.UpdatedAt.Format("2006-01-02T15:04:05Z07:00"), - } - if len(item.Segments) > 0 { - resp.Segments = item.Segments - } - if item.User != nil { - resp.Author = ConvertUserToResponse(item.User) - } - return resp +func ConvertTradeItemToDetailResponse(item *model.TradeItem, isFavorited bool) *TradeItemResponse { + return ConvertTradeItemToResponse(item, isFavorited) } \ No newline at end of file diff --git a/internal/dto/trade_dto.go b/internal/dto/trade_dto.go index d1fd1e4..6662299 100644 --- a/internal/dto/trade_dto.go +++ b/internal/dto/trade_dto.go @@ -34,26 +34,6 @@ type TradeItemResponse struct { UpdatedAt string `json:"updated_at"` } -type TradeItemDetailResponse struct { - ID string `json:"id"` - UserID string `json:"user_id"` - TradeType string `json:"trade_type"` - Category string `json:"category"` - Title string `json:"title"` - Content string `json:"content"` - Segments model.MessageSegments `json:"segments,omitempty"` - Images []TradeImageResponse `json:"images"` - Price *float64 `json:"price,omitempty"` - Condition string `json:"condition,omitempty"` - Status string `json:"status"` - ViewsCount int `json:"views_count"` - FavoritesCount int `json:"favorites_count"` - IsFavorited bool `json:"is_favorited"` - Author *UserResponse `json:"author"` - CreatedAt string `json:"created_at"` - UpdatedAt string `json:"updated_at"` -} - type CreateTradeItemRequest struct { TradeType string `json:"trade_type" binding:"required,oneof=sell buy"` Category string `json:"category" binding:"required,oneof=digital book clothing daily cosmetics sport ticket other"` diff --git a/internal/dto/user_converter.go b/internal/dto/user_converter.go index 3618ac6..3572a60 100644 --- a/internal/dto/user_converter.go +++ b/internal/dto/user_converter.go @@ -5,7 +5,20 @@ import ( "with_you/internal/pkg/utils" ) -// ==================== User 转换 ==================== +type UserResponseOption func(*UserResponse) + +func WithFollowing(isFollowing, isFollowingMe bool) UserResponseOption { + return func(r *UserResponse) { + r.IsFollowing = isFollowing + r.IsFollowingMe = isFollowingMe + } +} + +func WithPostsCount(count int) UserResponseOption { + return func(r *UserResponse) { + r.PostsCount = count + } +} // getAvatarOrDefault 获取头像URL，如果为空则返回在线头像生成服务的URL func getAvatarOrDefault(user *model.User) string { @@ -22,8 +35,7 @@ func publicVerificationFields(user *model.User) (model.UserIdentity, model.Verif return model.UserIdentityGeneral, model.VerificationStatusNone } -// ConvertUserToResponse 将User转换为UserResponse（公开信息，不包含敏感数据） -func ConvertUserToResponse(user *model.User) *UserResponse { +func baseUserResponse(user *model.User) *UserResponse { if user == nil { return nil } @@ -47,6 +59,18 @@ func ConvertUserToResponse(user *model.User) *UserResponse { } } +// ConvertUserToResponse 将User转换为UserResponse（公开信息，不包含敏感数据） +func ConvertUserToResponse(user *model.User, opts ...UserResponseOption) *UserResponse { + r := baseUserResponse(user) + if r == nil { + return nil + } + for _, opt := range opts { + opt(r) + } + return r +} + // ConvertUserToSelfResponse 将User转换为UserSelfResponse（本人信息，包含敏感数据） func ConvertUserToSelfResponse(user *model.User, postsCount int) *UserSelfResponse { if user == nil { @@ -75,151 +99,6 @@ func ConvertUserToSelfResponse(user *model.User, postsCount int) *UserSelfRespon } } -func convertPrivacySettingsToResponse(ps *model.PrivacySettings) *PrivacySettingsResponse { - if ps == nil { - return nil - } - return &PrivacySettingsResponse{ - FollowersVisibility: string(ps.FollowersVisibility), - FollowingVisibility: string(ps.FollowingVisibility), - PostsVisibility: string(ps.PostsVisibility), - FavoritesVisibility: string(ps.FavoritesVisibility), - } -} - -// ConvertUserToResponseWithFollowing 将User转换为UserResponse（包含关注状态，公开信息） -func ConvertUserToResponseWithFollowing(user *model.User, isFollowing bool) *UserResponse { - if user == nil { - return nil - } - identity, vs := publicVerificationFields(user) - return &UserResponse{ - ID: user.ID, - Username: user.Username, - Nickname: user.Nickname, - EmailVerified: user.EmailVerified, - Avatar: getAvatarOrDefault(user), - CoverURL: user.CoverURL, - Bio: user.Bio, - Website: user.Website, - Location: user.Location, - PostsCount: user.PostsCount, - FollowersCount: user.FollowersCount, - FollowingCount: user.FollowingCount, - IsFollowing: isFollowing, - IsFollowingMe: false, - CreatedAt: FormatTime(user.CreatedAt), - Identity: identity, - VerificationStatus: vs, - } -} - -// ConvertUserToResponseWithPostsCount 将User转换为UserResponse（使用实时计算的帖子数量，公开信息） -func ConvertUserToResponseWithPostsCount(user *model.User, postsCount int) *UserResponse { - if user == nil { - return nil - } - identity, vs := publicVerificationFields(user) - return &UserResponse{ - ID: user.ID, - Username: user.Username, - Nickname: user.Nickname, - EmailVerified: user.EmailVerified, - Avatar: getAvatarOrDefault(user), - CoverURL: user.CoverURL, - Bio: user.Bio, - Website: user.Website, - Location: user.Location, - PostsCount: postsCount, - FollowersCount: user.FollowersCount, - FollowingCount: user.FollowingCount, - CreatedAt: FormatTime(user.CreatedAt), - Identity: identity, - VerificationStatus: vs, - } -} - -// ConvertUserToResponseWithMutualFollow 将User转换为UserResponse（包含双向关注状态，公开信息） -func ConvertUserToResponseWithMutualFollow(user *model.User, isFollowing, isFollowingMe bool) *UserResponse { - if user == nil { - return nil - } - identity, vs := publicVerificationFields(user) - return &UserResponse{ - ID: user.ID, - Username: user.Username, - Nickname: user.Nickname, - EmailVerified: user.EmailVerified, - Avatar: getAvatarOrDefault(user), - CoverURL: user.CoverURL, - Bio: user.Bio, - Website: user.Website, - Location: user.Location, - PostsCount: user.PostsCount, - FollowersCount: user.FollowersCount, - FollowingCount: user.FollowingCount, - IsFollowing: isFollowing, - IsFollowingMe: isFollowingMe, - CreatedAt: FormatTime(user.CreatedAt), - Identity: identity, - VerificationStatus: vs, - } -} - -// ConvertUserToResponseWithMutualFollowAndPostsCount 将User转换为UserResponse（包含双向关注状态和实时计算的帖子数量，公开信息） -func ConvertUserToResponseWithMutualFollowAndPostsCount(user *model.User, isFollowing, isFollowingMe bool, postsCount int) *UserResponse { - if user == nil { - return nil - } - identity, vs := publicVerificationFields(user) - return &UserResponse{ - ID: user.ID, - Username: user.Username, - Nickname: user.Nickname, - EmailVerified: user.EmailVerified, - Avatar: getAvatarOrDefault(user), - CoverURL: user.CoverURL, - Bio: user.Bio, - Website: user.Website, - Location: user.Location, - PostsCount: postsCount, - FollowersCount: user.FollowersCount, - FollowingCount: user.FollowingCount, - IsFollowing: isFollowing, - IsFollowingMe: isFollowingMe, - CreatedAt: FormatTime(user.CreatedAt), - Identity: identity, - VerificationStatus: vs, - } -} - -// ConvertUserToDetailResponse 将User转换为UserDetailResponse（仅本人可见，包含敏感信息） -func ConvertUserToDetailResponse(user *model.User) *UserDetailResponse { - if user == nil { - return nil - } - return &UserDetailResponse{ - ID: user.ID, - Username: user.Username, - Nickname: user.Nickname, - Email: user.Email, - EmailVerified: user.EmailVerified, - Avatar: getAvatarOrDefault(user), - CoverURL: user.CoverURL, - Bio: user.Bio, - Website: user.Website, - Location: user.Location, - PostsCount: user.PostsCount, - FollowersCount: user.FollowersCount, - FollowingCount: user.FollowingCount, - IsVerified: user.IsVerified, - CreatedAt: FormatTime(user.CreatedAt), - PrivacySettings: convertPrivacySettingsToResponse(&user.PrivacySettings), - Identity: user.Identity, - VerificationStatus: user.VerificationStatus, - } -} - // ConvertUserToDetailResponseWithPostsCount 将User转换为UserDetailResponse（使用实时计算的帖子数量，仅本人可见） func ConvertUserToDetailResponseWithPostsCount(user *model.User, postsCount int) *UserDetailResponse { if user == nil { @@ -248,31 +127,28 @@ func ConvertUserToDetailResponseWithPostsCount(user *model.User, postsCount int) } } +func convertPrivacySettingsToResponse(ps *model.PrivacySettings) *PrivacySettingsResponse { + if ps == nil { + return nil + } + return &PrivacySettingsResponse{ + FollowersVisibility: string(ps.FollowersVisibility), + FollowingVisibility: string(ps.FollowingVisibility), + PostsVisibility: string(ps.PostsVisibility), + FavoritesVisibility: string(ps.FavoritesVisibility), + } +} + // ConvertUsersToResponse 将User列表转换为响应列表 -func ConvertUsersToResponse(users []*model.User) []*UserResponse { +func ConvertUsersToResponse(users []*model.User, opts ...UserResponseOption) []*UserResponse { result := make([]*UserResponse, 0, len(users)) for _, user := range users { - result = append(result, ConvertUserToResponse(user)) + result = append(result, ConvertUserToResponse(user, opts...)) } return result } -// ConvertUsersToResponseWithMutualFollow 将User列表转换为响应列表（包含双向关注状态） -// followingStatusMap: key是用户ID，value是[isFollowing, isFollowingMe] -func ConvertUsersToResponseWithMutualFollow(users []*model.User, followingStatusMap map[string][2]bool) []*UserResponse { - result := make([]*UserResponse, 0, len(users)) - for _, user := range users { - status, ok := followingStatusMap[user.ID] - if ok { - result = append(result, ConvertUserToResponseWithMutualFollow(user, status[0], status[1])) - } else { - result = append(result, ConvertUserToResponse(user)) - } - } - return result -} - -// ConvertUsersToResponseWithMutualFollowAndPostsCount 将User列表转换为响应列表（包含双向关注状态和实时计算的帖子数量） +// ConvertUsersToResponseWithMutualFollowAndPostsCount 将User列表转换为响应列表（包含双向关注状态和帖子数量） // followingStatusMap: key是用户ID，value是[isFollowing, isFollowingMe] // postsCountMap: key是用户ID，value是帖子数量 func ConvertUsersToResponseWithMutualFollowAndPostsCount(users []*model.User, followingStatusMap map[string][2]bool, postsCountMap map[string]int64) []*UserResponse { @@ -280,17 +156,15 @@ func ConvertUsersToResponseWithMutualFollowAndPostsCount(users []*model.User, fo for _, user := range users { status, hasStatus := followingStatusMap[user.ID] postsCount, hasPostsCount := postsCountMap[user.ID] - - // 如果没有帖子数量，使用数据库中的值 if !hasPostsCount { postsCount = int64(user.PostsCount) } - + var opts []UserResponseOption if hasStatus { - result = append(result, ConvertUserToResponseWithMutualFollowAndPostsCount(user, status[0], status[1], int(postsCount))) - } else { - result = append(result, ConvertUserToResponseWithPostsCount(user, int(postsCount))) + opts = append(opts, WithFollowing(status[0], status[1])) } + opts = append(opts, WithPostsCount(int(postsCount))) + result = append(result, ConvertUserToResponse(user, opts...)) } return result -} +} \ No newline at end of file diff --git a/internal/handler/message_handler.go b/internal/handler/message_handler.go index 1afd050..033e871 100644 --- a/internal/handler/message_handler.go +++ b/internal/handler/message_handler.go @@ -316,8 +316,7 @@ func (h *MessageHandler) SendMessage(c *gin.Context) { return } - // 直接使用 segments - msg, err := h.chatService.SendMessage(c.Request.Context(), userID, conversationID, req.Segments, req.ReplyToID) + msg, err := h.chatService.SendMessage(c.Request.Context(), userID, conversationID, req.Segments, req.ReplyToID, req.ClientMsgID) if err != nil { response.BadRequest(c, err.Error()) return @@ -359,7 +358,7 @@ func (h *MessageHandler) HandleSendMessage(c *gin.Context) { } // 发送消息 - msg, err := h.chatService.SendMessage(c.Request.Context(), userID, conversationID, params.Segments, params.ReplyToID) + msg, err := h.chatService.SendMessage(c.Request.Context(), userID, conversationID, params.Segments, params.ReplyToID, params.ClientMsgID) if err != nil { response.BadRequest(c, err.Error()) return diff --git a/internal/handler/post_handler.go b/internal/handler/post_handler.go index 740fc26..953fb7b 100644 --- a/internal/handler/post_handler.go +++ b/internal/handler/post_handler.go @@ -164,7 +164,7 @@ func (h *PostHandler) GetByID(c *gin.Context) { if currentUserID != "" { _, isFollowing, isFollowingMe, err := h.userService.GetUserByIDWithMutualFollowStatus(c.Request.Context(), post.UserID, currentUserID) if err == nil { - authorWithFollowStatus = dto.ConvertUserToResponseWithMutualFollow(post.User, isFollowing, isFollowingMe) + authorWithFollowStatus = dto.ConvertUserToResponse(post.User, dto.WithFollowing(isFollowing, isFollowingMe)) } else { authorWithFollowStatus = dto.ConvertUserToResponse(post.User) } diff --git a/internal/handler/user_handler.go b/internal/handler/user_handler.go index 40efd14..0568bde 100644 --- a/internal/handler/user_handler.go +++ b/internal/handler/user_handler.go @@ -307,7 +307,7 @@ func (h *UserHandler) GetUserByID(c *gin.Context) { } // 转换为响应格式，包含双向关注状态和实时计算的帖子数量 - userResponse := dto.ConvertUserToResponseWithMutualFollowAndPostsCount(user, isFollowing, isFollowingMe, int(postsCount)) + userResponse := dto.ConvertUserToResponse(user, dto.WithFollowing(isFollowing, isFollowingMe), dto.WithPostsCount(int(postsCount))) response.Success(c, userResponse) } diff --git a/internal/handler/ws_handler.go b/internal/handler/ws_handler.go index 253d1d9..006192a 100644 --- a/internal/handler/ws_handler.go +++ b/internal/handler/ws_handler.go @@ -147,7 +147,17 @@ func (h *WSHandler) HandleWebSocket(c *gin.Context) { } // 4. 注册客户端 - replayEvents := h.wsHub.Register(client) + replayEvents, regErr := h.wsHub.Register(client) + if regErr != nil { + zap.L().Warn("WebSocket registration rejected", + zap.String("user_id", userID), + zap.Error(regErr), + ) + conn.WriteMessage(websocket.CloseMessage, + []byte(`{"type":"error","payload":{"code":"connection_limit","message":"too many connections"}}`)) + conn.Close() + return + } defer h.wsHub.Unregister(client) zap.L().Info("WebSocket client connected", @@ -264,6 +274,11 @@ func (h *WSHandler) writePump(conn *websocket.Conn, client *ws.Client) { w, err := conn.NextWriter(websocket.TextMessage) if err != nil { + zap.L().Warn("WebSocket write error (NextWriter)", + zap.String("user_id", client.UserID), + zap.Uint64("client_id", client.ID), + zap.Error(err), + ) return } w.Write(message) @@ -276,6 +291,11 @@ func (h *WSHandler) writePump(conn *websocket.Conn, client *ws.Client) { } if err := w.Close(); err != nil { + zap.L().Warn("WebSocket write error (Close)", + zap.String("user_id", client.UserID), + zap.Uint64("client_id", client.ID), + zap.Error(err), + ) return } case <-ticker.C: @@ -349,6 +369,7 @@ func (h *WSHandler) handleChat(ctx context.Context, client *ws.Client, payload j DetailType string `json:"detail_type"` Segments model.MessageSegments `json:"segments"` ReplyToID *string `json:"reply_to_id,omitempty"` + ClientMsgID string `json:"client_msg_id,omitempty"` } if err := json.Unmarshal(payload, &req); err != nil { @@ -367,7 +388,7 @@ func (h *WSHandler) handleChat(ctx context.Context, client *ws.Client, payload j } // 发送消息 - message, err := h.chatService.SendMessage(ctx, client.UserID, req.ConversationID, req.Segments, req.ReplyToID) + message, err := h.chatService.SendMessage(ctx, client.UserID, req.ConversationID, req.Segments, req.ReplyToID, req.ClientMsgID) if err != nil { h.wsHub.SendError(client, "send_failed", err.Error()) return diff --git a/internal/model/audit_log.go b/internal/model/audit_log.go index 43dc27f..5fe83ba 100644 --- a/internal/model/audit_log.go +++ b/internal/model/audit_log.go @@ -3,7 +3,6 @@ package model import ( "time" - "github.com/google/uuid" "gorm.io/gorm" ) @@ -81,9 +80,7 @@ type AuditLog struct { // BeforeCreate 创建前生成UUID func (al *AuditLog) BeforeCreate(tx *gorm.DB) error { - if al.ID == "" { - al.ID = uuid.New().String() - } + SetUUIDIfEmpty(&al.ID) return nil } diff --git a/internal/model/call.go b/internal/model/call.go index 2296c5b..edbe055 100644 --- a/internal/model/call.go +++ b/internal/model/call.go @@ -1,12 +1,9 @@ package model import ( - "strconv" "time" "gorm.io/gorm" - - "with_you/internal/pkg/utils" ) // CallType 通话类型 @@ -60,14 +57,7 @@ type CallSession struct { // BeforeCreate 创建前生成雪花算法ID func (c *CallSession) BeforeCreate(tx *gorm.DB) error { - if c.ID == "" { - id, err := utils.GetSnowflake().GenerateID() - if err != nil { - return err - } - c.ID = strconv.FormatInt(id, 10) - } - return nil + return SetSnowflakeStringID(&c.ID) } func (CallSession) TableName() string { diff --git a/internal/model/channel.go b/internal/model/channel.go index 08e63db..eeccf8f 100644 --- a/internal/model/channel.go +++ b/internal/model/channel.go @@ -3,7 +3,6 @@ package model import ( "time" - "github.com/google/uuid" "gorm.io/gorm" ) @@ -20,9 +19,7 @@ type Channel struct { } func (c *Channel) BeforeCreate(tx *gorm.DB) error { - if c.ID == "" { - c.ID = uuid.New().String() - } + SetUUIDIfEmpty(&c.ID) return nil } diff --git a/internal/model/comment.go b/internal/model/comment.go index 5083dc3..71e17e0 100644 --- a/internal/model/comment.go +++ b/internal/model/comment.go @@ -3,7 +3,6 @@ package model import ( "time" - "github.com/google/uuid" "gorm.io/gorm" ) @@ -53,9 +52,7 @@ type Comment struct { // BeforeCreate 创建前生成UUID func (c *Comment) BeforeCreate(tx *gorm.DB) error { - if c.ID == "" { - c.ID = uuid.New().String() - } + SetUUIDIfEmpty(&c.ID) return nil } @@ -73,9 +70,7 @@ type CommentLike struct { // BeforeCreate 创建前生成UUID func (cl *CommentLike) BeforeCreate(tx *gorm.DB) error { - if cl.ID == "" { - cl.ID = uuid.New().String() - } + SetUUIDIfEmpty(&cl.ID) return nil } diff --git a/internal/model/conversation.go b/internal/model/conversation.go index 3e1170e..4b40342 100644 --- a/internal/model/conversation.go +++ b/internal/model/conversation.go @@ -1,12 +1,9 @@ package model import ( - "strconv" "time" "gorm.io/gorm" - - "with_you/internal/pkg/utils" ) // ConversationType 会话类型 @@ -36,14 +33,7 @@ type Conversation struct { // BeforeCreate 创建前生成雪花算法ID func (c *Conversation) BeforeCreate(tx *gorm.DB) error { - if c.ID == "" { - id, err := utils.GetSnowflake().GenerateID() - if err != nil { - return err - } - c.ID = strconv.FormatInt(id, 10) - } - return nil + return SetSnowflakeStringID(&c.ID) } func (Conversation) TableName() string { diff --git a/internal/model/device_token.go b/internal/model/device_token.go index 884640c..7b72987 100644 --- a/internal/model/device_token.go +++ b/internal/model/device_token.go @@ -4,8 +4,6 @@ import ( "time" "gorm.io/gorm" - - "with_you/internal/pkg/utils" ) // DeviceType 设备类型 @@ -41,14 +39,7 @@ type DeviceToken struct { // BeforeCreate 创建前生成雪花算法ID func (d *DeviceToken) BeforeCreate(tx *gorm.DB) error { - if d.ID == 0 { - id, err := utils.GetSnowflake().GenerateID() - if err != nil { - return err - } - d.ID = id - } - return nil + return SetSnowflakeInt64ID(&d.ID) } func (DeviceToken) TableName() string { diff --git a/internal/model/favorite.go b/internal/model/favorite.go index 521511c..b6f9144 100644 --- a/internal/model/favorite.go +++ b/internal/model/favorite.go @@ -3,7 +3,6 @@ package model import ( "time" - "github.com/google/uuid" "gorm.io/gorm" ) @@ -17,9 +16,7 @@ type Favorite struct { // BeforeCreate 创建前生成UUID func (f *Favorite) BeforeCreate(tx *gorm.DB) error { - if f.ID == "" { - f.ID = uuid.New().String() - } + SetUUIDIfEmpty(&f.ID) return nil } diff --git a/internal/model/follow.go b/internal/model/follow.go index c77a822..998cb89 100644 --- a/internal/model/follow.go +++ b/internal/model/follow.go @@ -3,7 +3,6 @@ package model import ( "time" - "github.com/google/uuid" "gorm.io/gorm" ) @@ -17,9 +16,7 @@ type Follow struct { // BeforeCreate 创建前生成UUID func (f *Follow) BeforeCreate(tx *gorm.DB) error { - if f.ID == "" { - f.ID = uuid.New().String() - } + SetUUIDIfEmpty(&f.ID) return nil } diff --git a/internal/model/group.go b/internal/model/group.go index c822cde..06b27dc 100644 --- a/internal/model/group.go +++ b/internal/model/group.go @@ -4,8 +4,6 @@ import ( "strconv" "time" - "with_you/internal/pkg/utils" - "gorm.io/gorm" ) @@ -35,14 +33,7 @@ type Group struct { // BeforeCreate 创建前生成雪花算法ID func (g *Group) BeforeCreate(tx *gorm.DB) error { - if g.ID == "" { - id, err := utils.GetSnowflake().GenerateID() - if err != nil { - return err - } - g.ID = strconv.FormatInt(id, 10) - } - return nil + return SetSnowflakeStringID(&g.ID) } // GetIDInt 获取数字类型的ID（用于比较） diff --git a/internal/model/group_announcement.go b/internal/model/group_announcement.go index 39a0b60..6a8c4db 100644 --- a/internal/model/group_announcement.go +++ b/internal/model/group_announcement.go @@ -1,11 +1,8 @@ package model import ( - "strconv" "time" - "with_you/internal/pkg/utils" - "gorm.io/gorm" ) @@ -22,14 +19,7 @@ type GroupAnnouncement struct { // BeforeCreate 创建前生成雪花算法ID func (ga *GroupAnnouncement) BeforeCreate(tx *gorm.DB) error { - if ga.ID == "" { - id, err := utils.GetSnowflake().GenerateID() - if err != nil { - return err - } - ga.ID = strconv.FormatInt(id, 10) - } - return nil + return SetSnowflakeStringID(&ga.ID) } // TableName 指定表名 diff --git a/internal/model/group_join_request.go b/internal/model/group_join_request.go index 688b558..a98d00f 100644 --- a/internal/model/group_join_request.go +++ b/internal/model/group_join_request.go @@ -1,11 +1,8 @@ package model import ( - "strconv" "time" - "with_you/internal/pkg/utils" - "gorm.io/gorm" ) @@ -44,14 +41,7 @@ type GroupJoinRequest struct { } func (r *GroupJoinRequest) BeforeCreate(tx *gorm.DB) error { - if r.ID == "" { - id, err := utils.GetSnowflake().GenerateID() - if err != nil { - return err - } - r.ID = strconv.FormatInt(id, 10) - } - return nil + return SetSnowflakeStringID(&r.ID) } func (GroupJoinRequest) TableName() string { diff --git a/internal/model/group_member.go b/internal/model/group_member.go index 83a389d..6e07b38 100644 --- a/internal/model/group_member.go +++ b/internal/model/group_member.go @@ -1,11 +1,8 @@ package model import ( - "strconv" "time" - "with_you/internal/pkg/utils" - "gorm.io/gorm" ) @@ -24,14 +21,7 @@ type GroupMember struct { // BeforeCreate 创建前生成雪花算法ID func (gm *GroupMember) BeforeCreate(tx *gorm.DB) error { - if gm.ID == "" { - id, err := utils.GetSnowflake().GenerateID() - if err != nil { - return err - } - gm.ID = strconv.FormatInt(id, 10) - } - return nil + return SetSnowflakeStringID(&gm.ID) } // TableName 指定表名 diff --git a/internal/model/id_helper.go b/internal/model/id_helper.go new file mode 100644 index 0000000..b54690d --- /dev/null +++ b/internal/model/id_helper.go @@ -0,0 +1,56 @@ +package model + +import ( + "strconv" + + "github.com/google/uuid" + "gorm.io/gorm" + + "with_you/internal/pkg/utils" +) + +func SetUUIDIfEmpty(id *string) { + if *id == "" { + *id = uuid.New().String() + } +} + +func SetSnowflakeStringID(id *string) error { + if *id == "" { + sid, err := utils.GetSnowflake().GenerateID() + if err != nil { + return err + } + *id = strconv.FormatInt(sid, 10) + } + return nil +} + +func SetSnowflakeInt64ID(id *int64) error { + if *id == 0 { + sid, err := utils.GetSnowflake().GenerateID() + if err != nil { + return err + } + *id = sid + } + return nil +} + +func GenerateUUID() string { + return uuid.New().String() +} + +func GenerateSnowflakeString() (string, error) { + id, err := utils.GetSnowflake().GenerateID() + if err != nil { + return "", err + } + return strconv.FormatInt(id, 10), nil +} + +func GenerateSnowflakeInt64() (int64, error) { + return utils.GetSnowflake().GenerateID() +} + +func nop(_ *gorm.DB) error { return nil } \ No newline at end of file diff --git a/internal/model/json_helper.go b/internal/model/json_helper.go new file mode 100644 index 0000000..d6f93a5 --- /dev/null +++ b/internal/model/json_helper.go @@ -0,0 +1,28 @@ +package model + +import ( + "database/sql/driver" + "encoding/json" +) + +// ScanJSON scans a database value (nil, []byte, or string) into any JSON-unmarshalable pointer. +func ScanJSON[T any](ptr *T, value any) error { + if value == nil { + return nil + } + var data []byte + switch v := value.(type) { + case []byte: + data = v + case string: + data = []byte(v) + default: + return nil + } + return json.Unmarshal(data, ptr) +} + +// ValueJSON marshals any value to JSON for database storage. +func ValueJSON[T any](val T) (driver.Value, error) { + return json.Marshal(val) +} \ No newline at end of file diff --git a/internal/model/like.go b/internal/model/like.go index 04fb9be..38b5d14 100644 --- a/internal/model/like.go +++ b/internal/model/like.go @@ -3,7 +3,6 @@ package model import ( "time" - "github.com/google/uuid" "gorm.io/gorm" ) @@ -17,9 +16,7 @@ type PostLike struct { // BeforeCreate 创建前生成UUID func (pl *PostLike) BeforeCreate(tx *gorm.DB) error { - if pl.ID == "" { - pl.ID = uuid.New().String() - } + SetUUIDIfEmpty(&pl.ID) return nil } diff --git a/internal/model/material.go b/internal/model/material.go index dcb8510..907fe69 100644 --- a/internal/model/material.go +++ b/internal/model/material.go @@ -2,10 +2,8 @@ package model import ( "database/sql/driver" - "encoding/json" "time" - "github.com/google/uuid" "gorm.io/gorm" ) @@ -44,9 +42,7 @@ type MaterialSubject struct { // BeforeCreate 创建前生成UUID func (s *MaterialSubject) BeforeCreate(tx *gorm.DB) error { - if s.ID == "" { - s.ID = uuid.New().String() - } + SetUUIDIfEmpty(&s.ID) return nil } @@ -62,7 +58,7 @@ func (a StringArray) Value() (driver.Value, error) { if a == nil { return nil, nil } - return json.Marshal(a) + return ValueJSON(a) } // Scan 实现 sql.Scanner 接口 @@ -71,16 +67,7 @@ func (a *StringArray) Scan(value any) error { *a = nil return nil } - var data []byte - switch v := value.(type) { - case []byte: - data = v - case string: - data = []byte(v) - default: - return nil - } - return json.Unmarshal(data, a) + return ScanJSON(a, value) } // MaterialFile 文件资料 @@ -107,9 +94,7 @@ type MaterialFile struct { // BeforeCreate 创建前生成UUID func (f *MaterialFile) BeforeCreate(tx *gorm.DB) error { - if f.ID == "" { - f.ID = uuid.New().String() - } + SetUUIDIfEmpty(&f.ID) return nil } diff --git a/internal/model/message.go b/internal/model/message.go index b135816..ba69e3d 100644 --- a/internal/model/message.go +++ b/internal/model/message.go @@ -3,15 +3,12 @@ import ( "database/sql/driver" "encoding/json" - "errors" - "strconv" "sync" "time" "gorm.io/gorm" "with_you/internal/pkg/crypto" - "with_you/internal/pkg/utils" ) // 系统消息相关常量 @@ -93,24 +90,12 @@ type ExtraData struct { // Value 实现driver.Valuer接口，用于数据库存储 func (e ExtraData) Value() (driver.Value, error) { - return json.Marshal(e) + return ValueJSON(e) } // Scan 实现sql.Scanner接口，用于数据库读取 func (e *ExtraData) Scan(value any) error { - if value == nil { - return nil - } - var data []byte - switch v := value.(type) { - case []byte: - data = v - case string: - data = []byte(v) - default: - return errors.New("type assertion to []byte or string failed") - } - return json.Unmarshal(data, e) + return ScanJSON(e, value) } // MessageSegmentData 单个消息段的数据 @@ -127,7 +112,7 @@ type MessageSegments []MessageSegment // Value 实现driver.Valuer接口，用于数据库存储 func (s MessageSegments) Value() (driver.Value, error) { - return json.Marshal(s) + return ValueJSON(s) } // Scan 实现sql.Scanner接口，用于数据库读取 @@ -136,16 +121,7 @@ func (s *MessageSegments) Scan(value any) error { *s = nil return nil } - var data []byte - switch v := value.(type) { - case []byte: - data = v - case string: - data = []byte(v) - default: - return errors.New("type assertion to []byte or string failed") - } - return json.Unmarshal(data, s) + return ScanJSON(s, value) } // Message 消息实体 @@ -188,12 +164,8 @@ func (m *Message) SenderIDStr() string { // BeforeCreate 创建前生成雪花算法ID并加密消息内容 func (m *Message) BeforeCreate(tx *gorm.DB) error { - if m.ID == "" { - id, err := utils.GetSnowflake().GenerateID() - if err != nil { - return err - } - m.ID = strconv.FormatInt(id, 10) + if err := SetSnowflakeStringID(&m.ID); err != nil { + return err } // 加密消息内容 diff --git a/internal/model/message_read.go b/internal/model/message_read.go deleted file mode 100644 index 00b3e24..0000000 --- a/internal/model/message_read.go +++ /dev/null @@ -1,19 +0,0 @@ -package model - -import ( - "time" -) - -// MessageRead 消息已读状态 -// 记录每个用户在每个会话中的已读位置 -type MessageRead struct { - ID uint `gorm:"primaryKey" json:"id"` - ConversationID int64 `gorm:"uniqueIndex:idx_conversation_user;not null" json:"conversation_id"` - UserID uint `gorm:"uniqueIndex:idx_conversation_user;not null" json:"user_id"` - LastReadSeq int64 `gorm:"not null" json:"last_read_seq"` // 已读到的seq位置 - UpdatedAt time.Time `json:"updated_at" gorm:"autoUpdateTime"` -} - -func (MessageRead) TableName() string { - return "message_reads" -} diff --git a/internal/model/notification.go b/internal/model/notification.go index 3248e74..8b77e7e 100644 --- a/internal/model/notification.go +++ b/internal/model/notification.go @@ -3,7 +3,6 @@ package model import ( "time" - "github.com/google/uuid" "gorm.io/gorm" ) @@ -42,9 +41,7 @@ type Notification struct { // BeforeCreate 创建前生成UUID func (n *Notification) BeforeCreate(tx *gorm.DB) error { - if n.ID == "" { - n.ID = uuid.New().String() - } + SetUUIDIfEmpty(&n.ID) return nil } diff --git a/internal/model/operation_log.go b/internal/model/operation_log.go index 6528393..230285e 100644 --- a/internal/model/operation_log.go +++ b/internal/model/operation_log.go @@ -3,7 +3,6 @@ package model import ( "time" - "github.com/google/uuid" "gorm.io/gorm" ) @@ -72,9 +71,7 @@ type OperationLog struct { // BeforeCreate 创建前生成TraceID func (ol *OperationLog) BeforeCreate(tx *gorm.DB) error { - if ol.TraceID == "" { - ol.TraceID = uuid.New().String() - } + SetUUIDIfEmpty(&ol.TraceID) if ol.OccurredAt.IsZero() { ol.OccurredAt = time.Now() } diff --git a/internal/model/post.go b/internal/model/post.go index 6f18124..cc85491 100644 --- a/internal/model/post.go +++ b/internal/model/post.go @@ -3,7 +3,6 @@ package model import ( "time" - "github.com/google/uuid" "gorm.io/gorm" ) @@ -66,9 +65,7 @@ type Post struct { // BeforeCreate 创建前生成UUID func (p *Post) BeforeCreate(tx *gorm.DB) error { - if p.ID == "" { - p.ID = uuid.New().String() - } + SetUUIDIfEmpty(&p.ID) return nil } @@ -94,9 +91,7 @@ type PostImage struct { // BeforeCreate 创建前生成UUID func (pi *PostImage) BeforeCreate(tx *gorm.DB) error { - if pi.ID == "" { - pi.ID = uuid.New().String() - } + SetUUIDIfEmpty(&pi.ID) return nil } diff --git a/internal/model/post_reference.go b/internal/model/post_reference.go index 124bd0e..0270210 100644 --- a/internal/model/post_reference.go +++ b/internal/model/post_reference.go @@ -3,7 +3,6 @@ package model import ( "time" - "github.com/google/uuid" "gorm.io/gorm" ) @@ -29,9 +28,7 @@ type PostReference struct { } func (pr *PostReference) BeforeCreate(tx *gorm.DB) error { - if pr.ID == "" { - pr.ID = uuid.New().String() - } + SetUUIDIfEmpty(&pr.ID) return nil } diff --git a/internal/model/push_record.go b/internal/model/push_record.go index b25f8b0..5de7a7c 100644 --- a/internal/model/push_record.go +++ b/internal/model/push_record.go @@ -4,8 +4,6 @@ import ( "time" "gorm.io/gorm" - - "with_you/internal/pkg/utils" ) // PushChannel 推送通道类型 @@ -66,14 +64,7 @@ type PushRecord struct { // BeforeCreate 创建前生成雪花算法ID func (r *PushRecord) BeforeCreate(tx *gorm.DB) error { - if r.ID == 0 { - id, err := utils.GetSnowflake().GenerateID() - if err != nil { - return err - } - r.ID = id - } - return nil + return SetSnowflakeInt64ID(&r.ID) } func (PushRecord) TableName() string { diff --git a/internal/model/report.go b/internal/model/report.go index 19e20ae..bfeccce 100644 --- a/internal/model/report.go +++ b/internal/model/report.go @@ -3,7 +3,6 @@ package model import ( "time" - "github.com/google/uuid" "gorm.io/gorm" ) @@ -60,9 +59,7 @@ type Report struct { // BeforeCreate 创建前生成UUID func (r *Report) BeforeCreate(tx *gorm.DB) error { - if r.ID == "" { - r.ID = uuid.New().String() - } + SetUUIDIfEmpty(&r.ID) return nil } diff --git a/internal/model/schedule_course.go b/internal/model/schedule_course.go index 1bf82cf..b1dc9fd 100644 --- a/internal/model/schedule_course.go +++ b/internal/model/schedule_course.go @@ -3,7 +3,6 @@ package model import ( "time" - "github.com/google/uuid" "gorm.io/gorm" ) @@ -24,9 +23,7 @@ type ScheduleCourse struct { } func (s *ScheduleCourse) BeforeCreate(tx *gorm.DB) error { - if s.ID == "" { - s.ID = uuid.New().String() - } + SetUUIDIfEmpty(&s.ID) return nil } diff --git a/internal/model/sensitive_word.go b/internal/model/sensitive_word.go index e533beb..f6689d9 100644 --- a/internal/model/sensitive_word.go +++ b/internal/model/sensitive_word.go @@ -3,7 +3,6 @@ package model import ( "time" - "github.com/google/uuid" "gorm.io/gorm" ) @@ -46,9 +45,7 @@ type SensitiveWord struct { // BeforeCreate 创建前生成UUID func (sw *SensitiveWord) BeforeCreate(tx *gorm.DB) error { - if sw.ID == "" { - sw.ID = uuid.New().String() - } + SetUUIDIfEmpty(&sw.ID) return nil } diff --git a/internal/model/sticker.go b/internal/model/sticker.go index f2963d3..e498b87 100644 --- a/internal/model/sticker.go +++ b/internal/model/sticker.go @@ -3,7 +3,6 @@ package model import ( "time" - "github.com/google/uuid" "gorm.io/gorm" ) @@ -26,8 +25,6 @@ func (UserSticker) TableName() string { // BeforeCreate 创建前生成UUID func (s *UserSticker) BeforeCreate(tx *gorm.DB) error { - if s.ID == "" { - s.ID = uuid.New().String() - } + SetUUIDIfEmpty(&s.ID) return nil } diff --git a/internal/model/system_notification.go b/internal/model/system_notification.go index 57b8aa8..1f0cf9e 100644 --- a/internal/model/system_notification.go +++ b/internal/model/system_notification.go @@ -2,13 +2,9 @@ import ( "database/sql/driver" - "encoding/json" - "errors" "time" "gorm.io/gorm" - - "with_you/internal/pkg/utils" ) // SystemNotificationType 系统通知类型 @@ -71,24 +67,12 @@ type SystemNotificationExtra struct { // Value 实现driver.Valuer接口 func (e SystemNotificationExtra) Value() (driver.Value, error) { - return json.Marshal(e) + return ValueJSON(e) } // Scan 实现sql.Scanner接口 func (e *SystemNotificationExtra) Scan(value any) error { - if value == nil { - return nil - } - var data []byte - switch v := value.(type) { - case []byte: - data = v - case string: - data = []byte(v) - default: - return errors.New("type assertion to []byte or string failed") - } - return json.Unmarshal(data, e) + return ScanJSON(e, value) } // SystemNotification 系统通知（独立表，与消息完全分离） @@ -113,14 +97,7 @@ type SystemNotification struct { // BeforeCreate 创建前生成雪花算法ID func (n *SystemNotification) BeforeCreate(tx *gorm.DB) error { - if n.ID == 0 { - id, err := utils.GetSnowflake().GenerateID() - if err != nil { - return err - } - n.ID = id - } - return nil + return SetSnowflakeInt64ID(&n.ID) } // TableName 指定表名 diff --git a/internal/model/trade.go b/internal/model/trade.go index 3aa966e..d972307 100644 --- a/internal/model/trade.go +++ b/internal/model/trade.go @@ -3,7 +3,6 @@ package model import ( "time" - "github.com/google/uuid" "gorm.io/gorm" ) @@ -57,9 +56,7 @@ type TradeItem struct { } func (t *TradeItem) BeforeCreate(tx *gorm.DB) error { - if t.ID == "" { - t.ID = uuid.New().String() - } + SetUUIDIfEmpty(&t.ID) return nil } @@ -81,9 +78,7 @@ type TradeImage struct { } func (ti *TradeImage) BeforeCreate(tx *gorm.DB) error { - if ti.ID == "" { - ti.ID = uuid.New().String() - } + SetUUIDIfEmpty(&ti.ID) return nil } @@ -99,9 +94,7 @@ type TradeFavorite struct { } func (f *TradeFavorite) BeforeCreate(tx *gorm.DB) error { - if f.ID == "" { - f.ID = uuid.New().String() - } + SetUUIDIfEmpty(&f.ID) return nil } diff --git a/internal/model/user.go b/internal/model/user.go index b465ad1..4e7765b 100644 --- a/internal/model/user.go +++ b/internal/model/user.go @@ -3,7 +3,6 @@ package model import ( "time" - "github.com/google/uuid" "gorm.io/gorm" ) @@ -106,9 +105,7 @@ type User struct { // BeforeCreate 创建前生成UUID func (u *User) BeforeCreate(tx *gorm.DB) error { - if u.ID == "" { - u.ID = uuid.New().String() - } + SetUUIDIfEmpty(&u.ID) return nil } diff --git a/internal/model/user_block.go b/internal/model/user_block.go index c4c9e62..9485f0f 100644 --- a/internal/model/user_block.go +++ b/internal/model/user_block.go @@ -3,7 +3,6 @@ package model import ( "time" - "github.com/google/uuid" "gorm.io/gorm" ) @@ -16,9 +15,7 @@ type UserBlock struct { } func (b *UserBlock) BeforeCreate(tx *gorm.DB) error { - if b.ID == "" { - b.ID = uuid.New().String() - } + SetUUIDIfEmpty(&b.ID) return nil } diff --git a/internal/model/user_profile_audit.go b/internal/model/user_profile_audit.go index 9811520..c072613 100644 --- a/internal/model/user_profile_audit.go +++ b/internal/model/user_profile_audit.go @@ -3,7 +3,6 @@ package model import ( "time" - "github.com/google/uuid" "gorm.io/gorm" ) @@ -40,9 +39,7 @@ type UserProfileAudit struct { } func (a *UserProfileAudit) BeforeCreate(tx *gorm.DB) error { - if a.ID == "" { - a.ID = uuid.New().String() - } + SetUUIDIfEmpty(&a.ID) return nil } diff --git a/internal/model/verification.go b/internal/model/verification.go index b4eb2a1..0708aa4 100644 --- a/internal/model/verification.go +++ b/internal/model/verification.go @@ -3,7 +3,6 @@ package model import ( "time" - "github.com/google/uuid" "gorm.io/gorm" ) @@ -29,9 +28,7 @@ type VerificationRecord struct { } func (v *VerificationRecord) BeforeCreate(tx *gorm.DB) error { - if v.ID == "" { - v.ID = uuid.New().String() - } + SetUUIDIfEmpty(&v.ID) return nil } diff --git a/internal/model/vote.go b/internal/model/vote.go index aff0983..2fe4a96 100644 --- a/internal/model/vote.go +++ b/internal/model/vote.go @@ -3,7 +3,6 @@ package model import ( "time" - "github.com/google/uuid" "gorm.io/gorm" ) @@ -20,9 +19,7 @@ type VoteOption struct { // BeforeCreate 创建前生成UUID func (vo *VoteOption) BeforeCreate(tx *gorm.DB) error { - if vo.ID == "" { - vo.ID = uuid.New().String() - } + SetUUIDIfEmpty(&vo.ID) return nil } @@ -41,9 +38,7 @@ type UserVote struct { // BeforeCreate 创建前生成UUID func (uv *UserVote) BeforeCreate(tx *gorm.DB) error { - if uv.ID == "" { - uv.ID = uuid.New().String() - } + SetUUIDIfEmpty(&uv.ID) return nil } diff --git a/internal/pkg/circuitbreaker/circuitbreaker.go b/internal/pkg/circuitbreaker/circuitbreaker.go new file mode 100644 index 0000000..5f9b726 --- /dev/null +++ b/internal/pkg/circuitbreaker/circuitbreaker.go @@ -0,0 +1,136 @@ +package circuitbreaker + +import ( + "sync" + "time" + + "go.uber.org/zap" +) + +type State int + +const ( + StateClosed State = iota + StateOpen + StateHalfOpen +) + +type Config struct { + FailureThreshold int // 连续失败次数阈值，超过则开启熔断 + SuccessThreshold int // 半开状态下连续成功次数，恢复到关闭 + Timeout time.Duration // 熔断开启后等待多久进入半开 + Name string // 熔断器名称，用于日志 +} + +func DefaultConfig(name string) Config { + return Config{ + FailureThreshold: 5, + SuccessThreshold: 3, + Timeout: 30 * time.Second, + Name: name, + } +} + +type Breaker struct { + cfg Config + mu sync.Mutex + state State + fails int + wins int + openAt time.Time +} + +func New(cfg Config) *Breaker { + return &Breaker{cfg: cfg} +} + +func (b *Breaker) Allow() bool { + b.mu.Lock() + defer b.mu.Unlock() + + switch b.state { + case StateClosed: + return true + case StateOpen: + if time.Since(b.openAt) > b.cfg.Timeout { + b.state = StateHalfOpen + b.wins = 0 + return true + } + return false + case StateHalfOpen: + return true + default: + return true + } +} + +func (b *Breaker) RecordSuccess() { + b.mu.Lock() + defer b.mu.Unlock() + + b.fails = 0 + if b.state == StateHalfOpen { + b.wins++ + if b.wins >= b.cfg.SuccessThreshold { + b.state = StateClosed + b.wins = 0 + zap.L().Info("circuit breaker closed", + zap.String("breaker", b.cfg.Name), + ) + } + } +} + +func (b *Breaker) RecordFailure() { + b.mu.Lock() + defer b.mu.Unlock() + + b.fails++ + if b.state == StateHalfOpen { + b.state = StateOpen + b.openAt = time.Now() + b.wins = 0 + zap.L().Warn("circuit breaker re-opened from half-open", + zap.String("breaker", b.cfg.Name), + ) + return + } + if b.fails >= b.cfg.FailureThreshold { + b.state = StateOpen + b.openAt = time.Now() + zap.L().Warn("circuit breaker opened", + zap.String("breaker", b.cfg.Name), + zap.Int("consecutive_failures", b.fails), + ) + } +} + +func (b *Breaker) State() State { + b.mu.Lock() + defer b.mu.Unlock() + return b.state +} + +func (b *Breaker) Execute(fn func() error) error { + if !b.Allow() { + zap.L().Debug("circuit breaker rejected request", + zap.String("breaker", b.cfg.Name), + ) + return &OpenError{Name: b.cfg.Name} + } + if err := fn(); err != nil { + b.RecordFailure() + return err + } + b.RecordSuccess() + return nil +} + +type OpenError struct { + Name string +} + +func (e *OpenError) Error() string { + return "circuit breaker '" + e.Name + "' is open" +} diff --git a/internal/pkg/jpush/client.go b/internal/pkg/jpush/client.go index 3c26cd3..2e32b68 100644 --- a/internal/pkg/jpush/client.go +++ b/internal/pkg/jpush/client.go @@ -13,6 +13,8 @@ import ( "strconv" "time" + "with_you/internal/pkg/circuitbreaker" + "go.uber.org/zap" ) @@ -29,6 +31,7 @@ type Client struct { httpClient *http.Client logger *zap.Logger lastRateLimit *RateLimitInfo + breaker *circuitbreaker.Breaker } func NewClient(appKey, masterSecret string, production bool, logger *zap.Logger) *Client { @@ -48,6 +51,12 @@ func NewClient(appKey, masterSecret string, production bool, logger *zap.Logger) production: production, httpClient: &http.Client{Timeout: 30 * time.Second, Transport: transport}, logger: logger, + breaker: circuitbreaker.New(circuitbreaker.Config{ + FailureThreshold: 5, + SuccessThreshold: 3, + Timeout: 30 * time.Second, + Name: "jpush", + }), } } @@ -259,7 +268,15 @@ func (c *Client) PushByRegistrationIDs(registrationIDs []string, notification *N }, } - result, _, err := c.Push(payload) + var result *PushResponse + err := c.breaker.Execute(func() error { + resp, _, pushErr := c.Push(payload) + if pushErr != nil { + return pushErr + } + result = resp + return nil + }) if err != nil { c.logger.Error("jpush push by registration_ids failed", zap.Int("count", len(registrationIDs)), diff --git a/internal/pkg/response/response.go b/internal/pkg/response/response.go index c677bf4..9e00d65 100644 --- a/internal/pkg/response/response.go +++ b/internal/pkg/response/response.go @@ -15,13 +15,6 @@ type Response struct { Data any `json:"data,omitempty"` } -// ResponseSnakeCase 统一响应结构(snake_case) -type ResponseSnakeCase struct { - Code int `json:"code"` - Message string `json:"message"` - Data any `json:"data,omitempty"` -} - // Success 成功响应 func Success(c *gin.Context, data any) { c.JSON(http.StatusOK, Response{ diff --git a/internal/pkg/ws/hub.go b/internal/pkg/ws/hub.go index 11dfaa6..deb2bc5 100644 --- a/internal/pkg/ws/hub.go +++ b/internal/pkg/ws/hub.go @@ -2,6 +2,7 @@ package ws import ( "encoding/json" + "errors" "sync" "sync/atomic" "time" @@ -10,8 +11,12 @@ import ( ) const ( - defaultUserBufferSize = 128 - maxReplayEvents = 200 + defaultUserBufferSize = 128 + maxReplayEvents = 200 + groupPushConcurrency = 64 + groupPushQueueSize = 512 + maxTotalConnections = 100000 + maxConnectionsPerUser = 10 ) // Event 表示一个WebSocket事件 @@ -31,6 +36,9 @@ type Client struct { Quit chan struct{} } +// ErrConnectionLimit 连接数限制错误 +var ErrConnectionLimit = errors.New("connection limit reached") + // Message 表示WebSocket消息 type Message struct { Type string `json:"type"` @@ -73,14 +81,15 @@ type ConnectHandler func(userID string, isFirstConnection bool) // Hub WebSocket连接管理器 type Hub struct { - seq uint64 + seq uint64 + connCount atomic.Int64 mu sync.RWMutex - clients map[string]map[uint64]*Client // userID -> clientID -> Client - history map[string][]Event // userID -> []Event (用于断线重连历史回放) - subscribers map[string]map[uint64]*subscriber // userID -> subscriberID -> subscriber - disconnectHandlers []DisconnectHandler // 断开连接事件处理器列表 - connectHandlers []ConnectHandler // 连接事件处理器列表 + clients map[string]map[uint64]*Client + history map[string][]Event + subscribers map[string]map[uint64]*subscriber + disconnectHandlers []DisconnectHandler + connectHandlers []ConnectHandler } // NewHub 创建WebSocket Hub @@ -99,11 +108,19 @@ func (h *Hub) NextID() uint64 { return atomic.AddUint64(&h.seq, 1) } -// Register 注册客户端连接 -func (h *Hub) Register(client *Client) []Event { +// Register 注册客户端连接，返回回放事件和可能的错误（连接数超限） +func (h *Hub) Register(client *Client) ([]Event, error) { + if h.connCount.Load() >= int64(maxTotalConnections) { + return nil, ErrConnectionLimit + } + h.mu.Lock() - // 检查是否是该用户的第一个连接（之前完全离线） + if len(h.clients[client.UserID]) >= maxConnectionsPerUser { + h.mu.Unlock() + return nil, ErrConnectionLimit + } + _, existed := h.clients[client.UserID] isFirstConnection := !existed @@ -111,8 +128,8 @@ func (h *Hub) Register(client *Client) []Event { h.clients[client.UserID] = make(map[uint64]*Client) } h.clients[client.UserID][client.ID] = client + h.connCount.Add(1) - // 获取历史事件用于回放 replay := make([]Event, 0) for _, e := range h.history[client.UserID] { replay = append(replay, e) @@ -122,15 +139,14 @@ func (h *Hub) Register(client *Client) []Event { zap.String("user_id", client.UserID), zap.Uint64("client_id", client.ID), zap.Int("total_clients", len(h.clients[client.UserID])), + zap.Int64("total_connections", h.connCount.Load()), zap.Bool("is_first_connection", isFirstConnection), ) - // 复制处理器列表避免锁竞争 handlers := make([]ConnectHandler, len(h.connectHandlers)) copy(handlers, h.connectHandlers) h.mu.Unlock() - // 异步调用连接处理器，避免阻塞 Register if len(handlers) > 0 { go func() { for _, handler := range handlers { @@ -139,7 +155,7 @@ func (h *Hub) Register(client *Client) []Event { }() } - return replay + return replay, nil } // Unregister 注销客户端连接 @@ -151,6 +167,7 @@ func (h *Hub) Unregister(client *Client) { if _, exists := userClients[client.ID]; exists { delete(userClients, client.ID) close(client.Quit) + h.connCount.Add(-1) remainingCount = len(userClients) if remainingCount == 0 { @@ -187,6 +204,20 @@ func (h *Hub) HasClients(userID string) bool { return len(h.clients[userID]) > 0 } +// FilterOnline 批量检查用户在线状态，一次加锁返回在线/离线列表 +func (h *Hub) FilterOnline(userIDs []string) (online []string, offline []string) { + h.mu.RLock() + defer h.mu.RUnlock() + for _, uid := range userIDs { + if len(h.clients[uid]) > 0 { + online = append(online, uid) + } else { + offline = append(offline, uid) + } + } + return +} + // OnDisconnect 注册断开连接事件处理器 // 当用户断开 WebSocket 连接时，会调用所有注册的处理器 // 处理器会收到用户ID和该用户剩余的连接数 @@ -224,10 +255,65 @@ func (h *Hub) PublishToUser(userID string, eventType string, payload any) Event return ev } -// PublishToUsers 向多个用户发送事件 +// PublishToUsers 向多个用户发送事件（共享一次序列化） func (h *Hub) PublishToUsers(userIDs []string, eventType string, payload any) { + if len(userIDs) == 0 { + return + } + + ev := Event{ + ID: h.NextID(), + Type: eventType, + TS: time.Now().UnixMilli(), + Payload: payload, + } + + msg := ResponseMessage{ + EventID: ev.ID, + Type: eventType, + TS: ev.TS, + Payload: payload, + } + data, err := json.Marshal(msg) + if err != nil { + zap.L().Error("Failed to marshal WebSocket message", + zap.String("event_type", eventType), + zap.Error(err), + ) + return + } + for _, uid := range userIDs { - h.PublishToUser(uid, eventType, payload) + h.publishPreMarshaled(uid, ev, data) + } +} + +// publishPreMarshaled 使用预序列化的数据发送事件 +func (h *Hub) publishPreMarshaled(userID string, ev Event, data []byte) { + h.mu.Lock() + history := append(h.history[userID], ev) + if len(history) > maxReplayEvents { + history = history[len(history)-maxReplayEvents:] + } + h.history[userID] = history + + targets := make([]*Client, 0, len(h.clients[userID])) + for _, c := range h.clients[userID] { + targets = append(targets, c) + } + h.mu.Unlock() + + for _, c := range targets { + select { + case <-c.Quit: + case c.Send <- data: + default: + zap.L().Warn("WebSocket client buffer full, closing slow client", + zap.String("user_id", userID), + zap.Uint64("client_id", c.ID), + ) + close(c.Quit) + } } } @@ -338,19 +424,16 @@ func (h *Hub) publish(userID string, ev Event) { return } - // 非阻塞发送，慢消费者丢弃消息 for _, c := range targets { select { case <-c.Quit: - // 客户端已断开 case c.Send <- data: - // 消息已发送 default: - // 发送缓冲区满，丢弃消息 - zap.L().Warn("WebSocket client buffer full, dropping message", + zap.L().Warn("WebSocket client buffer full, closing slow client", zap.String("user_id", userID), zap.Uint64("client_id", c.ID), ) + close(c.Quit) } } } @@ -387,6 +470,81 @@ func (h *Hub) Broadcast(eventType string, payload any) { h.PublishToUsers(userIDs, eventType, payload) } +// PublishToUsersConcurrent 并发推送消息给多个用户（适用于群聊等大列表场景） +// 使用有界并发控制，避免 goroutine 爆炸 +func (h *Hub) PublishToUsersConcurrent(userIDs []string, eventType string, payload any) { + if len(userIDs) == 0 { + return + } + + ev := Event{ + ID: h.NextID(), + Type: eventType, + TS: time.Now().UnixMilli(), + Payload: payload, + } + + msg := ResponseMessage{ + EventID: ev.ID, + Type: eventType, + TS: ev.TS, + Payload: payload, + } + data, err := json.Marshal(msg) + if err != nil { + zap.L().Error("Failed to marshal WebSocket message for batch push", + zap.String("event_type", eventType), + zap.Error(err), + ) + return + } + + sem := make(chan struct{}, groupPushConcurrency) + var wg sync.WaitGroup + wg.Add(len(userIDs)) + + for _, uid := range userIDs { + sem <- struct{}{} + go func(userID string) { + defer wg.Done() + defer func() { <-sem }() + + h.mu.RLock() + targets := make([]*Client, 0, len(h.clients[userID])) + for _, c := range h.clients[userID] { + targets = append(targets, c) + } + h.mu.RUnlock() + + for _, c := range targets { + select { + case <-c.Quit: + case c.Send <- data: + default: + zap.L().Warn("WebSocket client buffer full, closing slow client", + zap.String("user_id", userID), + zap.Uint64("client_id", c.ID), + ) + close(c.Quit) + } + } + }(uid) + } + + wg.Wait() + + // Store history for each user for reconnection replay + h.mu.Lock() + for _, uid := range userIDs { + history := append(h.history[uid], ev) + if len(history) > maxReplayEvents { + history = history[len(history)-maxReplayEvents:] + } + h.history[uid] = history + } + h.mu.Unlock() +} + // GetOnlineUsers 获取在线用户列表 func (h *Hub) GetOnlineUsers() []string { h.mu.RLock() @@ -406,6 +564,29 @@ func (h *Hub) GetOnlineCount() int { return len(h.clients) } +// GetConnectionCount 获取总连接数 +func (h *Hub) GetConnectionCount() int64 { + return h.connCount.Load() +} + +// CloseAllConnections 关闭所有WebSocket连接（用于优雅关闭） +func (h *Hub) CloseAllConnections() { + h.mu.Lock() + defer h.mu.Unlock() + + for _, userClients := range h.clients { + for _, c := range userClients { + select { + case <-c.Quit: + default: + close(c.Quit) + } + } + } + h.clients = make(map[string]map[uint64]*Client) + h.connCount.Store(0) +} + // Subscribe 订阅事件 func (h *Hub) Subscribe(userID string, afterID uint64) (chan Event, func(), []Event) { subID := h.NextID() diff --git a/internal/repository/group_join_request_repo.go b/internal/repository/group_join_request_repo.go index 528d432..f5a4932 100644 --- a/internal/repository/group_join_request_repo.go +++ b/internal/repository/group_join_request_repo.go @@ -1,6 +1,8 @@ package repository import ( + "errors" + "with_you/internal/model" "gorm.io/gorm" @@ -11,6 +13,7 @@ type GroupJoinRequestRepository interface { GetByFlag(flag string) (*model.GroupJoinRequest, error) Update(req *model.GroupJoinRequest) error GetPendingByGroupAndTarget(groupID, targetUserID string, reqType model.GroupJoinRequestType) (*model.GroupJoinRequest, error) + GetPendingByGroupAndTargets(groupID string, targetUserIDs []string, reqType model.GroupJoinRequestType) (map[string]bool, error) } type groupJoinRequestRepository struct { @@ -48,3 +51,25 @@ func (r *groupJoinRequestRepository) GetPendingByGroupAndTarget(groupID, targetU } return &req, nil } + +func (r *groupJoinRequestRepository) GetPendingByGroupAndTargets(groupID string, targetUserIDs []string, reqType model.GroupJoinRequestType) (map[string]bool, error) { + result := make(map[string]bool) + if len(targetUserIDs) == 0 { + return result, nil + } + var existingIDs []string + err := r.db.Model(&model.GroupJoinRequest{}). + Where("group_id = ? AND target_user_id IN ? AND request_type = ? AND status = ?", + groupID, targetUserIDs, reqType, model.GroupJoinRequestStatusPending). + Pluck("target_user_id", &existingIDs).Error + if err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return result, nil + } + return nil, err + } + for _, id := range existingIDs { + result[id] = true + } + return result, nil +} diff --git a/internal/repository/group_repo.go b/internal/repository/group_repo.go index ab8e0ba..b4a1cf8 100644 --- a/internal/repository/group_repo.go +++ b/internal/repository/group_repo.go @@ -21,6 +21,7 @@ type GroupRepository interface { // 群成员操作 AddMember(member *model.GroupMember) error + BatchAddMembers(groupID string, members []*model.GroupMember) error GetMember(groupID string, userID string) (*model.GroupMember, error) GetMembers(groupID string, page, pageSize int) ([]model.GroupMember, int64, error) UpdateMember(member *model.GroupMember) error @@ -53,6 +54,9 @@ type GroupRepository interface { GetUserGroupsByCursor(ctx context.Context, userID string, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.Group], error) GetMembersByCursor(ctx context.Context, groupID string, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.GroupMember], error) GetAnnouncementsByCursor(ctx context.Context, groupID string, req *cursor.PageRequest) (*cursor.CursorPageResult[*model.GroupAnnouncement], error) + + // 批量操作 + GetExistingMembers(groupID string, userIDs []string) (map[string]bool, error) } // GroupMemberWithUser 群成员带用户信息的结构 @@ -155,12 +159,24 @@ func (r *groupRepository) AddMember(member *model.GroupMember) error { if err := tx.Create(member).Error; err != nil { return err } - // 更新群组成员数量 return tx.Model(&model.Group{}).Where("id = ?", member.GroupID). Update("member_count", gorm.Expr("member_count + ?", 1)).Error }) } +func (r *groupRepository) BatchAddMembers(groupID string, members []*model.GroupMember) error { + if len(members) == 0 { + return nil + } + return r.db.Transaction(func(tx *gorm.DB) error { + if err := tx.CreateInBatches(members, 100).Error; err != nil { + return err + } + return tx.Model(&model.Group{}).Where("id = ?", groupID). + Update("member_count", gorm.Expr("member_count + ?", len(members))).Error + }) +} + // GetMember 获取群成员 func (r *groupRepository) GetMember(groupID string, userID string) (*model.GroupMember, error) { var member model.GroupMember diff --git a/internal/router/router.go b/internal/router/router.go index 86d366c..5254bfe 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -10,6 +10,46 @@ import ( "with_you/internal/service" ) +// RouterDeps 路由依赖 +type RouterDeps struct { + UserRepo repository.UserRepository + UserHandler *handler.UserHandler + PostHandler *handler.PostHandler + CommentHandler *handler.CommentHandler + MessageHandler *handler.MessageHandler + NotificationHandler *handler.NotificationHandler + UploadHandler *handler.UploadHandler + PushHandler *handler.PushHandler + SystemMessageHandler *handler.SystemMessageHandler + GroupHandler *handler.GroupHandler + StickerHandler *handler.StickerHandler + VoteHandler *handler.VoteHandler + ChannelHandler *handler.ChannelHandler + ScheduleHandler *handler.ScheduleHandler + RoleHandler *handler.RoleHandler + AdminUserHandler *handler.AdminUserHandler + AdminPostHandler *handler.AdminPostHandler + AdminCommentHandler *handler.AdminCommentHandler + AdminGroupHandler *handler.AdminGroupHandler + AdminDashboardHandler *handler.AdminDashboardHandler + AdminLogHandler *handler.AdminLogHandler + QRCodeHandler *handler.QRCodeHandler + MaterialHandler *handler.MaterialHandler + CallHandler *handler.CallHandler + ReportHandler *handler.ReportHandler + AdminReportHandler *handler.AdminReportHandler + VerificationHandler *handler.VerificationHandler + AdminVerificationHandler *handler.AdminVerificationHandler + AdminProfileAuditHandler *handler.AdminProfileAuditHandler + SetupHandler *handler.SetupHandler + TradeHandler *handler.TradeHandler + WSHandler *handler.WSHandler + JWTService *service.JWTService + LogService *service.LogService + ActivityService service.UserActivityService + CasbinService service.CasbinService +} + // Router 路由配置 type Router struct { engine *gin.Engine @@ -51,86 +91,47 @@ type Router struct { } // New 创建路由 -func New( - userRepo repository.UserRepository, - userHandler *handler.UserHandler, - postHandler *handler.PostHandler, - commentHandler *handler.CommentHandler, - messageHandler *handler.MessageHandler, - notificationHandler *handler.NotificationHandler, - uploadHandler *handler.UploadHandler, - jwtService *service.JWTService, - pushHandler *handler.PushHandler, - systemMessageHandler *handler.SystemMessageHandler, - groupHandler *handler.GroupHandler, - stickerHandler *handler.StickerHandler, - voteHandler *handler.VoteHandler, - channelHandler *handler.ChannelHandler, - scheduleHandler *handler.ScheduleHandler, - roleHandler *handler.RoleHandler, - adminUserHandler *handler.AdminUserHandler, - adminPostHandler *handler.AdminPostHandler, - adminCommentHandler *handler.AdminCommentHandler, - adminGroupHandler *handler.AdminGroupHandler, - adminDashboardHandler *handler.AdminDashboardHandler, - adminLogHandler *handler.AdminLogHandler, - qrcodeHandler *handler.QRCodeHandler, - materialHandler *handler.MaterialHandler, - callHandler *handler.CallHandler, - reportHandler *handler.ReportHandler, - adminReportHandler *handler.AdminReportHandler, - verificationHandler *handler.VerificationHandler, - adminVerificationHandler *handler.AdminVerificationHandler, - adminProfileAuditHandler *handler.AdminProfileAuditHandler, - setupHandler *handler.SetupHandler, - tradeHandler *handler.TradeHandler, - logService *service.LogService, - activityService service.UserActivityService, - casbinService service.CasbinService, - wsHandler *handler.WSHandler, -) *Router { - // 设置JWT服务 - userHandler.SetJWTService(jwtService) - // 设置活跃统计服务 - userHandler.SetActivityService(activityService) +func New(deps RouterDeps) *Router { + deps.UserHandler.SetJWTService(deps.JWTService) + deps.UserHandler.SetActivityService(deps.ActivityService) r := &Router{ engine: gin.Default(), - userRepo: userRepo, - userHandler: userHandler, - postHandler: postHandler, - commentHandler: commentHandler, - messageHandler: messageHandler, - notificationHandler: notificationHandler, - uploadHandler: uploadHandler, - pushHandler: pushHandler, - systemMessageHandler: systemMessageHandler, - groupHandler: groupHandler, - stickerHandler: stickerHandler, - voteHandler: voteHandler, - channelHandler: channelHandler, - scheduleHandler: scheduleHandler, - roleHandler: roleHandler, - adminUserHandler: adminUserHandler, - adminPostHandler: adminPostHandler, - adminCommentHandler: adminCommentHandler, - adminGroupHandler: adminGroupHandler, - adminDashboardHandler: adminDashboardHandler, - adminLogHandler: adminLogHandler, - qrcodeHandler: qrcodeHandler, - materialHandler: materialHandler, - callHandler: callHandler, - reportHandler: reportHandler, - adminReportHandler: adminReportHandler, - verificationHandler: verificationHandler, - adminVerificationHandler: adminVerificationHandler, - adminProfileAuditHandler: adminProfileAuditHandler, - setupHandler: setupHandler, - tradeHandler: tradeHandler, - logService: logService, - jwtService: jwtService, - casbinService: casbinService, - wsHandler: wsHandler, + userRepo: deps.UserRepo, + userHandler: deps.UserHandler, + postHandler: deps.PostHandler, + commentHandler: deps.CommentHandler, + messageHandler: deps.MessageHandler, + notificationHandler: deps.NotificationHandler, + uploadHandler: deps.UploadHandler, + pushHandler: deps.PushHandler, + systemMessageHandler: deps.SystemMessageHandler, + groupHandler: deps.GroupHandler, + stickerHandler: deps.StickerHandler, + voteHandler: deps.VoteHandler, + channelHandler: deps.ChannelHandler, + scheduleHandler: deps.ScheduleHandler, + roleHandler: deps.RoleHandler, + adminUserHandler: deps.AdminUserHandler, + adminPostHandler: deps.AdminPostHandler, + adminCommentHandler: deps.AdminCommentHandler, + adminGroupHandler: deps.AdminGroupHandler, + adminDashboardHandler: deps.AdminDashboardHandler, + adminLogHandler: deps.AdminLogHandler, + qrcodeHandler: deps.QRCodeHandler, + materialHandler: deps.MaterialHandler, + callHandler: deps.CallHandler, + reportHandler: deps.ReportHandler, + adminReportHandler: deps.AdminReportHandler, + verificationHandler: deps.VerificationHandler, + adminVerificationHandler: deps.AdminVerificationHandler, + adminProfileAuditHandler: deps.AdminProfileAuditHandler, + setupHandler: deps.SetupHandler, + tradeHandler: deps.TradeHandler, + logService: deps.LogService, + jwtService: deps.JWTService, + casbinService: deps.CasbinService, + wsHandler: deps.WSHandler, } r.setupRoutes() diff --git a/internal/service/chat_service.go b/internal/service/chat_service.go index 71cb518..fbf577d 100644 --- a/internal/service/chat_service.go +++ b/internal/service/chat_service.go @@ -20,6 +20,9 @@ import ( // 撤回消息的时间限制（2分钟） const RecallMessageTimeout = 2 * time.Minute +// 消息幂等性 TTL：24 小时内同一 client_msg_id 不会重复发送 +const messageIdempotentTTL = 24 * time.Hour + // ChatService 聊天服务接口 type ChatService interface { // 会话管理 @@ -31,7 +34,7 @@ type ChatService interface { SetConversationNotificationMuted(ctx context.Context, conversationID string, userID string, notificationMuted bool) error // 消息操作 - SendMessage(ctx context.Context, senderID string, conversationID string, segments model.MessageSegments, replyToID *string) (*model.Message, error) + SendMessage(ctx context.Context, senderID string, conversationID string, segments model.MessageSegments, replyToID *string, clientMsgID string) (*model.Message, error) GetMessages(ctx context.Context, conversationID string, userID string, page, pageSize int) ([]*model.Message, int64, error) GetMessagesAfterSeq(ctx context.Context, conversationID string, userID string, afterSeq int64, limit int) ([]*model.Message, error) GetMessagesBeforeSeq(ctx context.Context, conversationID string, userID string, beforeSeq int64, limit int) ([]*model.Message, error) @@ -57,15 +60,15 @@ type ChatService interface { // chatServiceImpl 聊天服务实现 type chatServiceImpl struct { - repo repository.MessageRepository - userRepo repository.UserRepository - sensitive SensitiveService - wsHub *ws.Hub - pushSvc PushService + repo repository.MessageRepository + userRepo repository.UserRepository + sensitive SensitiveService + wsHub *ws.Hub + pushSvc PushService + cache cache.Cache - // 缓存相关字段 conversationCache *cache.ConversationCache - uploadService *UploadService // 校验聊天图片 URL 来源 + uploadService *UploadService } // NewChatService 创建聊天服务 @@ -96,6 +99,7 @@ func NewChatService( sensitive: sensitive, wsHub: wsHub, pushSvc: pushSvc, + cache: cacheBackend, conversationCache: conversationCache, uploadService: uploadService, } @@ -245,9 +249,16 @@ func (s *chatServiceImpl) SetConversationNotificationMuted(ctx context.Context, return nil } -// SendMessage 发送消息（使用 segments） -func (s *chatServiceImpl) SendMessage(ctx context.Context, senderID string, conversationID string, segments model.MessageSegments, replyToID *string) (*model.Message, error) { - // 首先验证会话是否存在 +func (s *chatServiceImpl) SendMessage(ctx context.Context, senderID string, conversationID string, segments model.MessageSegments, replyToID *string, clientMsgID string) (*model.Message, error) { + if clientMsgID != "" && s.cache != nil { + idemKey := cache.MessageIdempotentKey(senderID, clientMsgID) + if cachedMsgID, ok := cache.GetTyped[string](s.cache, idemKey); ok && cachedMsgID != "" { + if existing, err := s.repo.GetMessageByID(cachedMsgID); err == nil && existing != nil { + return existing, nil + } + } + } + conv, err := s.getConversation(ctx, conversationID) if err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { @@ -332,7 +343,6 @@ func (s *chatServiceImpl) SendMessage(ctx context.Context, senderID string, conv s.conversationCache.InvalidateMessagePages(conversationID) } - // 异步写入缓存 go func() { if err := s.cacheMessage(context.Background(), conversationID, message); err != nil { zap.L().Warn("async cache message failed", @@ -344,12 +354,15 @@ func (s *chatServiceImpl) SendMessage(ctx context.Context, senderID string, conv } }() + if clientMsgID != "" && s.cache != nil { + s.cache.Set(cache.MessageIdempotentKey(senderID, clientMsgID), message.ID, messageIdempotentTTL) + } + // 获取会话中的参与者并发送消息 participants, err := s.getParticipants(ctx, conversationID) if err == nil { targetIDs := make([]string, 0, len(participants)) for _, p := range participants { - // 私聊场景下，发送者已经从 HTTP 响应拿到消息，避免再通过 WebSocket 回推导致本端重复展示。 if conv.Type == model.ConversationTypePrivate && p.UserID == senderID { continue } @@ -359,18 +372,21 @@ func (s *chatServiceImpl) SendMessage(ctx context.Context, senderID string, conv if conv.Type == model.ConversationTypeGroup { detailType = "group" } - s.publishToUsers(targetIDs, "chat_message", map[string]any{ - "detail_type": detailType, - "message": dto.ConvertMessageToResponse(message), - }) + if len(targetIDs) > 20 && conv.Type == model.ConversationTypeGroup { + s.wsHub.PublishToUsersConcurrent(targetIDs, "chat_message", map[string]any{ + "detail_type": detailType, + "message": dto.ConvertMessageToResponse(message), + }) + } else { + s.publishToUsers(targetIDs, "chat_message", map[string]any{ + "detail_type": detailType, + "message": dto.ConvertMessageToResponse(message), + }) + } for _, p := range participants { if p.UserID == senderID { continue } - // 失效未读数缓存 - if s.conversationCache != nil { - s.conversationCache.InvalidateUnreadCount(p.UserID, conversationID) - } if totalUnread, uErr := s.repo.GetAllUnreadCount(p.UserID); uErr == nil { s.publishToUsers([]string{p.UserID}, "conversation_unread", map[string]any{ "conversation_id": conversationID, @@ -380,14 +396,22 @@ func (s *chatServiceImpl) SendMessage(ctx context.Context, senderID string, conv } } - // 失效会话列表缓存 if s.conversationCache != nil { + keys := make([]string, 0, len(participants)*2) + for _, p := range participants { + keys = append(keys, + cache.UnreadDetailKey(p.UserID, conversationID), + cache.UnreadConversationKey(p.UserID), + ) + } + if len(keys) > 0 { + s.cache.DeleteBatch(keys) + } for _, p := range participants { s.conversationCache.InvalidateConversationList(p.UserID) } } - // 对离线用户通过 JPush 推送聊天消息通知 if s.pushSvc != nil && len(participants) > 0 { sender := ChatMessageSender{ID: senderID} if senderUser, sErr := s.userRepo.GetByID(senderID); sErr == nil { @@ -399,26 +423,37 @@ func (s *chatServiceImpl) SendMessage(ctx context.Context, senderID string, conv if conv.Type == model.ConversationTypeGroup && conv.Group != nil { convName = conv.Group.Name } + allTargetIDs := make([]string, 0, len(participants)) for _, p := range participants { if p.UserID == senderID { continue } - go func(userID string, sender ChatMessageSender) { - if pushErr := s.pushSvc.PushChatMessage(context.Background(), userID, conversationID, &sender, convType, convName, message); pushErr != nil { - zap.L().Debug("push chat message skipped or failed", - zap.String("userID", userID), - zap.String("conversationID", conversationID), - zap.Error(pushErr), - ) + allTargetIDs = append(allTargetIDs, p.UserID) + } + var offlineTargetIDs []string + if s.wsHub != nil && len(allTargetIDs) > 0 { + _, offlineTargetIDs = s.wsHub.FilterOnline(allTargetIDs) + } else { + offlineTargetIDs = allTargetIDs + } + if len(offlineTargetIDs) > 0 { + go func(ids []string, sender ChatMessageSender) { + for _, uid := range ids { + if pushErr := s.pushSvc.PushChatMessage(context.Background(), uid, conversationID, &sender, convType, convName, message); pushErr != nil { + zap.L().Debug("push chat message failed", + zap.String("userID", uid), + zap.String("conversationID", conversationID), + zap.Error(pushErr), + ) + } } - }(p.UserID, sender) + }(offlineTargetIDs, sender) } } return message, nil } -// getConversation 获取会话（优先使用缓存） func (s *chatServiceImpl) getConversation(ctx context.Context, conversationID string) (*model.Conversation, error) { if s.conversationCache != nil { return s.conversationCache.GetConversation(ctx, conversationID) diff --git a/internal/service/trade_service.go b/internal/service/trade_service.go index 75074ca..da2ada2 100644 --- a/internal/service/trade_service.go +++ b/internal/service/trade_service.go @@ -17,7 +17,7 @@ import ( type TradeService interface { Create(ctx context.Context, userID string, req dto.CreateTradeItemRequest) (*model.TradeItem, error) - GetByID(ctx context.Context, id string, currentUserID *string) (*dto.TradeItemDetailResponse, error) + GetByID(ctx context.Context, id string, currentUserID *string) (*dto.TradeItemResponse, error) Update(ctx context.Context, userID string, id string, req dto.UpdateTradeItemRequest) error Delete(ctx context.Context, userID string, id string) error UpdateStatus(ctx context.Context, userID string, id string, status model.TradeItemStatus) error @@ -62,7 +62,7 @@ func (s *tradeService) Create(ctx context.Context, userID string, req dto.Create return item, nil } -func (s *tradeService) GetByID(ctx context.Context, id string, currentUserID *string) (*dto.TradeItemDetailResponse, error) { +func (s *tradeService) GetByID(ctx context.Context, id string, currentUserID *string) (*dto.TradeItemResponse, error) { item, err := s.tradeRepo.GetByID(ctx, id) if err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { @@ -81,7 +81,7 @@ func (s *tradeService) GetByID(ctx context.Context, id string, currentUserID *st _ = s.tradeRepo.IncrementViews(ctx, id) - return dto.ConvertTradeItemToDetailResponse(item, isFavorited), nil + return dto.ConvertTradeItemToResponse(item, isFavorited), nil } func (s *tradeService) Update(ctx context.Context, userID string, id string, req dto.UpdateTradeItemRequest) error {