From 7f683ce1505599656ebd9bca59e9ee33e9592c55 Mon Sep 17 00:00:00 2001 From: zhr Date: Tue, 22 Sep 2026 11:01:56 +0800 Subject: [PATCH] =?UTF-8?q?feat:=201=E3=80=81optimmize=20deploy=202?= =?UTF-8?q?=E3=80=81message=20insert?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .dockerignore | 23 +++---- Dockerfile | 29 ++++----- Makefile | 13 +++- README.md | 6 +- deploy/docker/config.docker.yaml | 4 ++ deploy/docker/deploy.sh | 62 ++++++++++++------- deploy/docker/docker-compose.yaml | 3 - deploy/nginx/README.md | 2 +- internal/module/auth/logic/jwt.go | 22 +++---- internal/module/group/model/group.go | 4 +- .../module/message/model/group_message.go | 6 +- .../module/message/model/private_message.go | 4 +- internal/module/message/model/repo.go | 6 +- internal/module/message/model/row.go | 21 ++++--- internal/module/message/model/row_test.go | 36 ++++++++++- internal/pkg/errs/apperror.go | 19 ++++++ internal/pkg/errs/apperror_test.go | 18 ++++++ internal/transport/http/auth.go | 13 ++++ 18 files changed, 195 insertions(+), 96 deletions(-) diff --git a/.dockerignore b/.dockerignore index d73ac35..d1f7171 100644 --- a/.dockerignore +++ b/.dockerignore @@ -1,16 +1,7 @@ -.git -.cursor -.idea -.vscode -bin -logs -vendor -examples -docs -*.md -*.test -*.out -coverage.out -.env -certs -deploy/nginx +** +!bin/ +!bin/linux_amd64/ +!bin/linux_amd64/spark-server +!bin/linux_amd64/spark-migrate +!migrations/ +!migrations/** diff --git a/Dockerfile b/Dockerfile index bf48796..edce814 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,25 +1,22 @@ -# syntax=docker/dockerfile:1 -FROM golang:1.26-bookworm AS build -WORKDIR /src -COPY go.mod go.sum ./ -RUN go mod download -COPY . . -RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -trimpath -ldflags="-s -w" -o /out/spark-server ./cmd/server \ - && CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -trimpath -ldflags="-s -w" -o /out/spark-migrate ./cmd/migrate - -FROM alpine:3.20 -RUN apk add --no-cache ca-certificates tzdata wget \ - && adduser -D -H -u 10001 spark +# Runtime only. Binaries are built on the developer machine (see Makefile `image`). +FROM swr.cn-north-4.myhuaweicloud.com/ddn-k8s/gcr.io/iguazio/alpine:3.20 +RUN apk add --no-cache ca-certificates tzdata \ + && adduser -D -H -u 10001 spark \ + && mkdir -p /usr/share/zoneinfo/Asia \ + && cp /usr/share/zoneinfo/Asia/Shanghai /tmp/Shanghai \ + && apk del tzdata \ + && mkdir -p /usr/share/zoneinfo/Asia \ + && mv /tmp/Shanghai /usr/share/zoneinfo/Asia/Shanghai \ + && ln -sf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime ENV TZ=Asia/Shanghai WORKDIR /app -COPY --from=build /out/spark-server /usr/local/bin/spark-server -COPY --from=build /out/spark-migrate /usr/local/bin/spark-migrate +COPY bin/linux_amd64/spark-server /usr/local/bin/spark-server +COPY bin/linux_amd64/spark-migrate /usr/local/bin/spark-migrate COPY migrations /app/migrations -COPY deploy/docker/config.docker.yaml /app/configs/config.yaml RUN mkdir -p /app/logs && chown -R spark:spark /app USER spark EXPOSE 8080 9090 HEALTHCHECK --interval=15s --timeout=3s --start-period=10s --retries=3 \ - CMD wget -qO- http://127.0.0.1:8080/healthz >/dev/null || exit 1 + CMD wget -qO- http://127.0.0.1:8080/healthz >/dev/null || exit 1 ENTRYPOINT ["/usr/local/bin/spark-server"] CMD ["-config", "/app/configs/config.yaml"] diff --git a/Makefile b/Makefile index 2e88c6a..b282c14 100644 --- a/Makefile +++ b/Makefile @@ -1,17 +1,28 @@ GO ?= go MODULE := spark-server BIN := bin/spark-server +LINUX_DIR := bin/linux_amd64 +IMAGE ?= spark-server:latest +LDFLAGS := -s -w -buildid= LINT ?= golangci-lint CONFIG ?= configs/config.yaml LOGIC_COVER_MIN ?= 65 -.PHONY: all build test lint fmt vet tidy migrate start run deploy +.PHONY: all build image test lint fmt vet tidy migrate start run deploy all: lint test build build: $(GO) build -o $(BIN) ./cmd/server +# Cross-compile for ECS and pack a runtime-only amd64 image (no Go toolchain in the image). +image: + @command -v docker >/dev/null || { echo "docker is required for make image" >&2; exit 1; } + mkdir -p $(LINUX_DIR) + CGO_ENABLED=0 GOOS=linux GOARCH=amd64 $(GO) build -trimpath -ldflags="$(LDFLAGS)" -o $(LINUX_DIR)/spark-server ./cmd/server + CGO_ENABLED=0 GOOS=linux GOARCH=amd64 $(GO) build -trimpath -ldflags="$(LDFLAGS)" -o $(LINUX_DIR)/spark-migrate ./cmd/migrate + docker build --platform linux/amd64 -t $(IMAGE) . + # All unit tests, then each module logic package must be >= LOGIC_COVER_MIN. test: $(GO) test ./... diff --git a/README.md b/README.md index 0c976e0..1764941 100644 --- a/README.md +++ b/README.md @@ -284,16 +284,16 @@ curl -sS http://127.0.0.1:8080/readyz ## 部署(ECS) -`make deploy` → `deploy/docker/deploy.sh`。默认 `root@47.236.66.9`,密钥 `~/.ssh/spark-ecs`。 +`make deploy` → 本机交叉编译 linux/amd64、打运行时镜像、再 `docker save` 到 ECS `docker load`。默认 `root@47.236.66.9`,密钥 `~/.ssh/spark-ecs`。需要本机有 `go` 和 `docker`。只打镜像:`make image`。 主机目录: | 路径 | 内容 | | --- | --- | -| `/data/spark-server/app` | 构建源码 | +| `/data/spark-server/compose/docker-compose.yaml` | 只跑已 load 的 `spark-server:latest` | | `/data/spark-server/configs/config.yaml` | 运行配置(来自 `deploy/docker/config.docker.yaml`) | | `/data/spark-server/logs` | 进程日志 | -| `/data/spark-server/images/spark-server.tar.gz` | `docker save` | +| `/data/spark-server/images/spark-server.tar.gz` | 本机 `docker save` 上传 | 容器加入已有 `shell_default` 网络(mysql / redis / rustfs)。发布: diff --git a/deploy/docker/config.docker.yaml b/deploy/docker/config.docker.yaml index 2952694..92b59fc 100644 --- a/deploy/docker/config.docker.yaml +++ b/deploy/docker/config.docker.yaml @@ -57,6 +57,10 @@ jwt: denylist_prefix: "denylist:" refresh_prefix: "refresh:" +# Must match Aliyun wyrs.admin.jwt-secret / WYRS_ADMIN_JWT_SECRET (popadmin login). +admin: + jwt_secret: "wyrs-admin-jwt-secret-change-in-production" + taskpool: size: 256 queue: 1024 diff --git a/deploy/docker/deploy.sh b/deploy/docker/deploy.sh index 111b198..1879497 100755 --- a/deploy/docker/deploy.sh +++ b/deploy/docker/deploy.sh @@ -1,38 +1,54 @@ #!/usr/bin/env bash -# Deploy spark-server to the ECS host and join mysql/redis/rustfs on shell_default. +# Build linux/amd64 binaries + runtime image locally, then load them on the ECS host. # Host layout: -# /data/spark-server/app source (docker build context) -# /data/spark-server/configs config.yaml -# /data/spark-server/logs process logs -# /data/spark-server/images docker save of spark-server:latest +# /data/spark-server/compose docker-compose.yaml +# /data/spark-server/configs config.yaml +# /data/spark-server/logs process logs +# /data/spark-server/images spark-server.tar.gz set -euo pipefail ROOT="$(cd "$(dirname "$0")/../.." && pwd)" HOST="${SPARK_SSH_HOST:-root@47.236.66.9}" KEY="${SPARK_SSH_KEY:-$HOME/.ssh/spark-ecs}" DATA="${SPARK_DATA_DIR:-/data/spark-server}" -REMOTE="${SPARK_REMOTE_DIR:-$DATA/app}" +IMAGE="${SPARK_IMAGE:-spark-server:latest}" +TAR="${SPARK_IMAGE_TAR:-$ROOT/bin/spark-server.tar.gz}" SSH=(ssh -i "$KEY" -o BatchMode=yes -o StrictHostKeyChecking=accept-new "$HOST") +SCP=(scp -i "$KEY" -o BatchMode=yes -o StrictHostKeyChecking=accept-new) -"${SSH[@]}" "mkdir -p $DATA/app $DATA/configs $DATA/logs $DATA/images" +need() { + command -v "$1" >/dev/null || { + echo "need $1 on this machine" >&2 + exit 1 + } +} -rsync -az --delete \ - --exclude '.git/' \ - --exclude 'bin/' \ - --exclude 'logs/' \ - --exclude 'vendor/' \ - --exclude 'examples/' \ - --exclude '.cursor/' \ - --exclude '.idea/' \ - --exclude '.vscode/' \ - -e "ssh -i $KEY -o BatchMode=yes" \ - "$ROOT/" "$HOST:$REMOTE/" +need go +need docker +need rsync +need ssh +need scp -"${SSH[@]}" "cp $REMOTE/deploy/docker/config.docker.yaml $DATA/configs/config.yaml && chown -R 10001:10001 $DATA/logs" -"${SSH[@]}" "cd $REMOTE/deploy/docker && docker compose -f docker-compose.yaml build" -"${SSH[@]}" "docker save spark-server:latest | gzip -c > $DATA/images/spark-server.tar.gz" -"${SSH[@]}" "cd $REMOTE/deploy/docker && docker compose -f docker-compose.yaml up -d spark-server" +echo "==> local linux/amd64 binaries + image" +make -C "$ROOT" image +mkdir -p "$(dirname "$TAR")" +docker save "$IMAGE" | gzip -c >"$TAR" +echo "image $(docker image inspect "$IMAGE" --format '{{.Size}}') bytes, tar $(wc -c <"$TAR") bytes" + +echo "==> upload" +"${SSH[@]}" "mkdir -p $DATA/compose $DATA/configs $DATA/logs $DATA/images" +"${SCP[@]}" "$TAR" "$HOST:$DATA/images/spark-server.tar.gz" +"${SCP[@]}" "$ROOT/deploy/docker/docker-compose.yaml" "$HOST:$DATA/compose/docker-compose.yaml" +"${SCP[@]}" "$ROOT/deploy/docker/config.docker.yaml" "$HOST:$DATA/configs/config.yaml" +rsync -az -e "ssh -i $KEY -o BatchMode=yes" \ + "$ROOT/deploy/nginx/host/" "$HOST:/tmp/spark-nginx-host/" + +echo "==> load and start" +"${SSH[@]}" "chown -R 10001:10001 $DATA/logs && docker load -i $DATA/images/spark-server.tar.gz" +# Replace any stale spark-server container (e.g. old docker run) so compose can recreate. +"${SSH[@]}" "docker rm -f spark-server 2>/dev/null || true" +"${SSH[@]}" "cd $DATA/compose && docker compose -f docker-compose.yaml up -d --force-recreate spark-server" "${SSH[@]}" "docker exec spark-server /usr/local/bin/spark-migrate -config /app/configs/config.yaml -dir /app/migrations" -"${SSH[@]}" "cp $REMOTE/deploy/nginx/host/spark-sni.conf /etc/nginx/stream.d/spark-sni.conf && if [ -f /etc/letsencrypt/live/sparkr.wyrs365.com/fullchain.pem ]; then cp $REMOTE/deploy/nginx/host/sparkr.conf /etc/nginx/conf.d/sparkr.conf; fi && nginx -t && systemctl reload nginx" +"${SSH[@]}" "cp /tmp/spark-nginx-host/spark-sni.conf /etc/nginx/stream.d/spark-sni.conf && if [ -f /etc/letsencrypt/live/sparkr.wyrs365.com/fullchain.pem ]; then cp /tmp/spark-nginx-host/sparkr.conf /etc/nginx/conf.d/sparkr.conf; fi && nginx -t && systemctl reload nginx" "${SSH[@]}" "docker ps --filter name=spark-server --format 'table {{.Names}}\t{{.Status}}\t{{.Ports}}'" "${SSH[@]}" "curl -fsS http://127.0.0.1:18080/healthz && echo && curl -fsS http://127.0.0.1:18080/readyz && echo" diff --git a/deploy/docker/docker-compose.yaml b/deploy/docker/docker-compose.yaml index 03a7e6b..2251bc0 100644 --- a/deploy/docker/docker-compose.yaml +++ b/deploy/docker/docker-compose.yaml @@ -3,9 +3,6 @@ # Image/config/logs live under /data/spark-server on the ECS host. services: spark-server: - build: - context: ../.. - dockerfile: Dockerfile image: spark-server:latest container_name: spark-server restart: unless-stopped diff --git a/deploy/nginx/README.md b/deploy/nginx/README.md index a17f11c..45ec3f1 100644 --- a/deploy/nginx/README.md +++ b/deploy/nginx/README.md @@ -24,7 +24,7 @@ spark-server (Docker): - HTTP `127.0.0.1:18080` → container `:8080` - TCP `127.0.0.1:9090` with `proxy_protocol: true` → container `:9090` (host `:9000` is rustfs S3; bind `127.0.0.1:9000`, not public) -- Image / config / logs: `/data/spark-server/{images,configs,logs}` +- Image / config / logs: `/data/spark-server/{images,compose,configs,logs}`(镜像在本机打包后 `docker load`) - Public security group: 443 only - Clients upload/download at `https://sparkr.wyrs365.com` (path-style). spark-server talks to `rustfs:9000` on the Docker network and signs URLs with `object_storage.public_endpoint`. - Unsigned GET is allowed only for `/perm/avatar/` and `/perm/group/` (rewritten to `/spark/...`). Chat and other prefixes stay SigV4. diff --git a/internal/module/auth/logic/jwt.go b/internal/module/auth/logic/jwt.go index 3c3d58c..cf28be8 100644 --- a/internal/module/auth/logic/jwt.go +++ b/internal/module/auth/logic/jwt.go @@ -164,7 +164,7 @@ func (j *JWT) sign(userID int64, deviceID uint64, use string, ttl time.Duration) func (j *JWT) replacePair(ctx context.Context, userID int64, deviceID uint64, accessJTI, refreshJTI string) error { old, err := j.pairs.Get(ctx, userID, deviceID) if err != nil { - return errs.ErrAvailable.WithError("jwt-refresh", err) + return errs.FromDependencyError("jwt-refresh", err) } if old.Access != "" { _ = j.denylist.Add(ctx, old.Access, j.ttl) @@ -173,7 +173,7 @@ func (j *JWT) replacePair(ctx context.Context, userID int64, deviceID uint64, ac _ = j.denylist.Add(ctx, old.Refresh, j.refreshTTL) } if err := j.pairs.Put(ctx, userID, deviceID, TokenPair{Access: accessJTI, Refresh: refreshJTI}, j.refreshTTL); err != nil { - return errs.ErrAvailable.WithError("jwt-refresh", err) + return errs.FromDependencyError("jwt-refresh", err) } return nil } @@ -188,7 +188,7 @@ func (j *JWT) Refresh(ctx context.Context, refreshToken string) (*authentity.Iss } denied, err := j.denylist.Has(ctx, c.ID) if err != nil { - return nil, errs.ErrAvailable.WithError("jwt-denylist", err) + return nil, errs.FromDependencyError("jwt-denylist", err) } if denied { return nil, errs.ErrUnauthorized @@ -199,7 +199,7 @@ func (j *JWT) Refresh(ctx context.Context, refreshToken string) (*authentity.Iss } stored, err := j.pairs.Get(ctx, uid, c.DeviceID) if err != nil { - return nil, errs.ErrAvailable.WithError("jwt-refresh", err) + return nil, errs.FromDependencyError("jwt-refresh", err) } if stored.Refresh == "" || stored.Refresh != c.ID { return nil, errs.ErrUnauthorized @@ -212,13 +212,13 @@ func (j *JWT) Logout(ctx context.Context, userID int64, deviceID uint64, accessJ remain := accessExp.Sub(j.clock()) if remain > 0 { if err := j.denylist.Add(ctx, accessJTI, remain); err != nil { - return errs.ErrAvailable.WithError("jwt-denylist", err) + return errs.FromDependencyError("jwt-denylist", err) } } } old, err := j.pairs.Get(ctx, userID, deviceID) if err != nil { - return errs.ErrAvailable.WithError("jwt-refresh", err) + return errs.FromDependencyError("jwt-refresh", err) } if old.Access != "" && old.Access != accessJTI { _ = j.denylist.Add(ctx, old.Access, j.ttl) @@ -227,7 +227,7 @@ func (j *JWT) Logout(ctx context.Context, userID int64, deviceID uint64, accessJ _ = j.denylist.Add(ctx, old.Refresh, j.refreshTTL) } if err := j.pairs.Delete(ctx, userID, deviceID); err != nil { - return errs.ErrAvailable.WithError("jwt-refresh", err) + return errs.FromDependencyError("jwt-refresh", err) } return nil } @@ -238,7 +238,7 @@ func (j *JWT) LogoutAll(ctx context.Context, userID int64) error { } rows, err := j.pairs.ListByUser(ctx, userID) if err != nil { - return errs.ErrAvailable.WithError("jwt-refresh", err) + return errs.FromDependencyError("jwt-refresh", err) } for i := range rows { row := rows[i] @@ -249,7 +249,7 @@ func (j *JWT) LogoutAll(ctx context.Context, userID int64) error { _ = j.denylist.Add(ctx, row.Pair.Refresh, j.refreshTTL) } if err := j.pairs.Delete(ctx, userID, row.DeviceID); err != nil { - return errs.ErrAvailable.WithError("jwt-refresh", err) + return errs.FromDependencyError("jwt-refresh", err) } } return nil @@ -262,7 +262,7 @@ func (j *JWT) Authenticate(ctx context.Context, token string) (*authentity.Princ } denied, err := j.denylist.Has(ctx, c.ID) if err != nil { - return nil, errs.ErrAvailable.WithError("jwt-denylist", err) + return nil, errs.FromDependencyError("jwt-denylist", err) } if denied { return nil, errs.ErrUnauthorized @@ -294,7 +294,7 @@ func (j *JWT) Revoke(ctx context.Context, token string) error { return nil } if err := j.denylist.Add(ctx, c.ID, remain); err != nil { - return errs.ErrAvailable.WithError("jwt-denylist", err) + return errs.FromDependencyError("jwt-denylist", err) } return nil } diff --git a/internal/module/group/model/group.go b/internal/module/group/model/group.go index e58ded1..f2d7309 100644 --- a/internal/module/group/model/group.go +++ b/internal/module/group/model/group.go @@ -98,7 +98,7 @@ func (m *Repo) ListByUser(ctx context.Context, userID int64, limit int) ([]*enti } limit = entity.ClampLimit(limit) var rows []Group - err = db.Table("groups AS g").Select("g.*"). + err = db.Table("`groups` AS g").Select("g.*"). Joins("INNER JOIN group_members m ON m.group_id = g.id"). Where("m.user_id = ? AND g.status = ?", userID, entity.StatusActive). Order("g.updated_at DESC").Limit(limit).Find(&rows).Error @@ -123,7 +123,7 @@ func (m *Repo) ListAllActiveByUser(ctx context.Context, userID int64) ([]*entity return nil, err } var rows []Group - err = db.Table("groups AS g").Select("g.*"). + err = db.Table("`groups` AS g").Select("g.*"). Joins("INNER JOIN group_members m ON m.group_id = g.id"). Where("m.user_id = ? AND g.status = ?", userID, entity.StatusActive). Order("g.updated_at DESC").Limit(purgeListCap).Find(&rows).Error diff --git a/internal/module/message/model/group_message.go b/internal/module/message/model/group_message.go index b46bfd7..52fc154 100644 --- a/internal/module/message/model/group_message.go +++ b/internal/module/message/model/group_message.go @@ -4,8 +4,8 @@ import "spark-server/internal/module/message/entity" // GroupMessage maps table group_message. type GroupMessage struct { - messageRow - GroupID int64 `gorm:"column:group_id;comment:群号 groups.id,便于按群查询"` + MessageRow `gorm:"embedded"` + GroupID int64 `gorm:"column:group_id;comment:群号 groups.id,便于按群查询"` } func (GroupMessage) TableName() string { return tableGroupMessage } @@ -14,7 +14,7 @@ func (r *GroupMessage) entity() *entity.Message { if r == nil { return nil } - m := r.messageRow.entity(entity.KindGroup) + m := r.MessageRow.entity(entity.KindGroup) m.GroupID = r.GroupID return m } diff --git a/internal/module/message/model/private_message.go b/internal/module/message/model/private_message.go index 8840928..694d9d8 100644 --- a/internal/module/message/model/private_message.go +++ b/internal/module/message/model/private_message.go @@ -4,7 +4,7 @@ import "spark-server/internal/module/message/entity" // PrivateMessage maps table private_message. type PrivateMessage struct { - messageRow + MessageRow `gorm:"embedded"` ReceiverID int64 `gorm:"column:receiver_id;comment:接收者 user_id;系统消息为 0"` } @@ -14,7 +14,7 @@ func (r *PrivateMessage) entity() *entity.Message { if r == nil { return nil } - m := r.messageRow.entity(entity.KindPrivate) + m := r.MessageRow.entity(entity.KindPrivate) m.ReceiverID = r.ReceiverID return m } diff --git a/internal/module/message/model/repo.go b/internal/module/message/model/repo.go index 58f7c94..6f707d8 100644 --- a/internal/module/message/model/repo.go +++ b/internal/module/message/model/repo.go @@ -168,7 +168,7 @@ func (m *Repo) CreateAndBump(ctx context.Context, message *entity.Message) error }) } -func stampCreatedAt(message *entity.Message, row *messageRow) { +func stampCreatedAt(message *entity.Message, row *MessageRow) { if message.CreatedAtMS > 0 { row.CreatedAt = time.UnixMilli(message.CreatedAtMS) return @@ -177,7 +177,7 @@ func stampCreatedAt(message *entity.Message, row *messageRow) { message.CreatedAtMS = row.CreatedAt.UnixMilli() } -func (m *Repo) insertBody(db *gorm.DB, message *entity.Message, row messageRow) error { +func (m *Repo) insertBody(db *gorm.DB, message *entity.Message, row MessageRow) error { if message.Kind == entity.KindGroup { if err := m.fillGroupID(db, message); err != nil { return err @@ -197,7 +197,7 @@ func (m *Repo) fillGroupID(db *gorm.DB, message *entity.Message) error { return nil } var gid int64 - if err := db.Table("groups").Select("id").Where("conversation_id = ?", message.ConversationID).Scan(&gid).Error; err != nil { + if err := db.Table("`groups`").Select("id").Where("conversation_id = ?", message.ConversationID).Scan(&gid).Error; err != nil { return mapDBErr(err) } if gid <= 0 { diff --git a/internal/module/message/model/row.go b/internal/module/message/model/row.go index 4aca80d..f39e5fd 100644 --- a/internal/module/message/model/row.go +++ b/internal/module/message/model/row.go @@ -11,8 +11,9 @@ const ( tableGroupMessage = "group_message" ) -type messageRow struct { - ID int64 `gorm:"column:id;primaryKey;comment:Snowflake MsgId,主键、排序键、同步游标"` +// MessageRow is shared columns for private_message and group_message. Must be exported so GORM can embed it. +type MessageRow struct { + ID int64 `gorm:"column:id;primaryKey;autoIncrement:false;comment:Snowflake MsgId,主键、排序键、同步游标"` ConversationID int64 `gorm:"column:conversation_id;comment:所属会话 id"` SenderID int64 `gorm:"column:sender_id;comment:发送者 user_id"` ClientMsgID string `gorm:"column:client_msg_id;comment:客户端消息 id,与 sender+会话 组成幂等键"` @@ -22,7 +23,7 @@ type messageRow struct { CreatedAt time.Time `gorm:"column:created_at;comment:服务端入库时间"` } -func (r *messageRow) entity(kind entity.Kind) *entity.Message { +func (r *MessageRow) entity(kind entity.Kind) *entity.Message { if r == nil { return nil } @@ -39,11 +40,11 @@ func (r *messageRow) entity(kind entity.Kind) *entity.Message { } } -func rowFrom(m *entity.Message) messageRow { +func rowFrom(m *entity.Message) MessageRow { if m == nil { - return messageRow{} + return MessageRow{} } - return messageRow{ + return MessageRow{ ID: m.ID, ConversationID: m.ConversationID, SenderID: m.SenderID, @@ -54,12 +55,12 @@ func rowFrom(m *entity.Message) messageRow { } } -func privateRow(m *entity.Message, base messageRow) PrivateMessage { - return PrivateMessage{messageRow: base, ReceiverID: m.ReceiverID} +func privateRow(m *entity.Message, base MessageRow) PrivateMessage { + return PrivateMessage{MessageRow: base, ReceiverID: m.ReceiverID} } -func groupRow(m *entity.Message, base messageRow) GroupMessage { - return GroupMessage{messageRow: base, GroupID: m.GroupID} +func groupRow(m *entity.Message, base MessageRow) GroupMessage { + return GroupMessage{MessageRow: base, GroupID: m.GroupID} } func tableOf(kind entity.Kind) string { diff --git a/internal/module/message/model/row_test.go b/internal/module/message/model/row_test.go index 596a33b..096d404 100644 --- a/internal/module/message/model/row_test.go +++ b/internal/module/message/model/row_test.go @@ -1,8 +1,12 @@ package model import ( + "reflect" + "sync" "testing" + "gorm.io/gorm/schema" + "spark-server/internal/module/message/entity" ) @@ -19,12 +23,40 @@ func TestTableOf(t *testing.T) { } func TestQueryKeyRows(t *testing.T) { - priv := PrivateMessage{messageRow: messageRow{ID: 1, SenderID: 2}, ReceiverID: 3} + priv := PrivateMessage{MessageRow: MessageRow{ID: 1, SenderID: 2}, ReceiverID: 3} if got := priv.entity(); got.ReceiverID != 3 || got.Kind != entity.KindPrivate { t.Fatalf("%+v", got) } - grp := GroupMessage{messageRow: messageRow{ID: 4, SenderID: 5}, GroupID: 9} + grp := GroupMessage{MessageRow: MessageRow{ID: 4, SenderID: 5}, GroupID: 9} if got := grp.entity(); got.GroupID != 9 || got.Kind != entity.KindGroup { t.Fatalf("%+v", got) } } + +func TestPrivateMessageEmbedsMessageRowForGORM(t *testing.T) { + f, ok := reflect.TypeOf(PrivateMessage{}).FieldByName("MessageRow") + if !ok || f.Tag.Get("gorm") != "embedded" { + t.Fatalf("want gorm embedded on MessageRow, got %+v", f) + } + rec := PrivateMessage{MessageRow: MessageRow{ID: 99, ConversationID: 1, SenderID: 2}, ReceiverID: 3} + if rec.ID != 99 || rec.ConversationID != 1 { + t.Fatalf("embedded fields not promoted %+v", rec) + } +} + +func TestPrivateMessageGORMSchemaColumns(t *testing.T) { + var cache sync.Map + s, err := schema.Parse(&PrivateMessage{}, &cache, schema.NamingStrategy{}) + if err != nil { + t.Fatal(err) + } + want := []string{"id", "conversation_id", "sender_id", "client_msg_id", "message_type", "content", "status", "created_at", "receiver_id"} + if len(s.DBNames) != len(want) { + t.Fatalf("columns %v want %v", s.DBNames, want) + } + for i, name := range want { + if s.DBNames[i] != name { + t.Fatalf("col[%d] got %s want %s (all %v)", i, s.DBNames[i], name, s.DBNames) + } + } +} diff --git a/internal/pkg/errs/apperror.go b/internal/pkg/errs/apperror.go index daac668..10599e6 100644 --- a/internal/pkg/errs/apperror.go +++ b/internal/pkg/errs/apperror.go @@ -106,3 +106,22 @@ func FromRPCError(serviceName string, err error) *AppError { return ErrInner.WithInnerMsg(fmt.Sprintf("[%s] unexpected error: %v", serviceName, err)) } } + +// FromDependencyError maps Redis/MySQL (or similar) errors for HTTP/TCP auth and token flows. +// context.Canceled usually means the client disconnected before the dependency returned. +func FromDependencyError(serviceName string, err error) *AppError { + if err == nil { + return nil + } + if app, ok := AsAppError(err); ok { + return app + } + switch { + case errors.Is(err, context.Canceled): + return ErrClosed.WithInnerMsg(fmt.Sprintf("[%s] canceled", serviceName)) + case errors.Is(err, context.DeadlineExceeded): + return ErrTimeout.WithInnerMsg(fmt.Sprintf("[%s] timeout", serviceName)) + default: + return ErrAvailable.WithError(serviceName, err) + } +} diff --git a/internal/pkg/errs/apperror_test.go b/internal/pkg/errs/apperror_test.go index fcc91e6..00bfb13 100644 --- a/internal/pkg/errs/apperror_test.go +++ b/internal/pkg/errs/apperror_test.go @@ -37,3 +37,21 @@ func TestFromRPCError(t *testing.T) { t.Fatal("inner") } } + +func TestFromDependencyError(t *testing.T) { + if FromDependencyError("s", nil) != nil { + t.Fatal("nil") + } + if got := FromDependencyError("s", ErrBusy); !errors.Is(got, ErrBusy) { + t.Fatal("passthrough") + } + if got := FromDependencyError("jwt-denylist", context.Canceled); !errors.Is(got, ErrClosed) { + t.Fatal("canceled") + } + if got := FromDependencyError("jwt-denylist", context.DeadlineExceeded); !errors.Is(got, ErrTimeout) { + t.Fatal("timeout") + } + if got := FromDependencyError("jwt-denylist", errors.New("conn refused")); !errors.Is(got, ErrAvailable) { + t.Fatal("available") + } +} diff --git a/internal/transport/http/auth.go b/internal/transport/http/auth.go index 789a283..05b3f01 100644 --- a/internal/transport/http/auth.go +++ b/internal/transport/http/auth.go @@ -1,6 +1,8 @@ package httpx import ( + "context" + "errors" "strconv" "strings" @@ -117,5 +119,16 @@ func matchDeviceHeader(raw string, want uint64) error { func abortAuth(c *gin.Context, err error) { c.Abort() + if clientGone(c, err) { + return + } JSONError(c, err) } + +// clientGone is true when the peer closed the connection; writing a JSON body is pointless. +func clientGone(c *gin.Context, err error) bool { + if errors.Is(c.Request.Context().Err(), context.Canceled) { + return true + } + return errors.Is(err, errs.ErrClosed) || errors.Is(err, context.Canceled) +}