333 lines
15 KiB
Go
333 lines
15 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"crypto/sha256"
|
||
"encoding/hex"
|
||
"errors"
|
||
"strings"
|
||
"time"
|
||
|
||
roomeventsv1 "hyapp.local/api/proto/events/room/v1"
|
||
roomv1 "hyapp.local/api/proto/room/v1"
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/giftlimits"
|
||
"hyapp/pkg/roomid"
|
||
"hyapp/pkg/xerr"
|
||
"hyapp/services/room-service/internal/room/command"
|
||
"hyapp/services/room-service/internal/room/outbox"
|
||
"hyapp/services/room-service/internal/room/rank"
|
||
"hyapp/services/room-service/internal/room/state"
|
||
)
|
||
|
||
// AuthorizeRoomMediaUpload 在 COS PutObject 之前执行房间 owner/presence、治理态和 VIP 权益门禁。
|
||
// 返回的 prefix 是 gateway 生成对象 key 的唯一可信目录，客户端不能自行指定 key。
|
||
func (s *Service) AuthorizeRoomMediaUpload(ctx context.Context, req *roomv1.AuthorizeRoomMediaUploadRequest) (*roomv1.AuthorizeRoomMediaUploadResponse, error) {
|
||
if req == nil || req.GetMeta() == nil {
|
||
return nil, xerr.New(xerr.InvalidArgument, "request is required")
|
||
}
|
||
ctx = contextFromMeta(ctx, req.GetMeta())
|
||
roomID := strings.TrimSpace(req.GetMeta().GetRoomId())
|
||
actorUserID := req.GetMeta().GetActorUserId()
|
||
purpose := strings.ToLower(strings.TrimSpace(req.GetPurpose()))
|
||
if !roomid.ValidStringID(roomID) || actorUserID <= 0 || !giftlimits.ValidCommandID(req.GetMeta().GetCommandId()) ||
|
||
(purpose != roomMediaPurposeBackground && purpose != roomMediaPurposeImage) {
|
||
return nil, xerr.New(xerr.InvalidArgument, "room media upload request is invalid")
|
||
}
|
||
media := roomMediaFromProto(req.GetMedia())
|
||
if err := validateRoomMediaProperties(purpose, media); err != nil {
|
||
return nil, err
|
||
}
|
||
if media.URL != "" || media.ObjectKey != "" || media.Status != roomMediaStatusActive {
|
||
return nil, xerr.New(xerr.InvalidArgument, "room media authorization requires pre-upload metadata")
|
||
}
|
||
guard, ok := s.repository.(RoomMediaUploadGuardStore)
|
||
if !ok {
|
||
return nil, xerr.New(xerr.Unavailable, "room media upload guard is unavailable")
|
||
}
|
||
payloadSHA256, err := roomMediaUploadPayloadSHA256(media)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
registration := RoomMediaUploadRegistration{
|
||
AppCode: appcode.FromContext(ctx), RoomID: roomID, CommandID: strings.TrimSpace(req.GetMeta().GetCommandId()),
|
||
ActorUserID: actorUserID, Purpose: purpose, MediaPayloadSHA256: payloadSHA256, Media: media,
|
||
}
|
||
registration.ExpectedObjectKey = roomMediaExpectedObjectKey(
|
||
registration.AppCode, actorUserID, roomID, purpose, registration.CommandID, media,
|
||
)
|
||
var authorizedReplay *RoomMediaUploadRegistration
|
||
if existing, exists, readErr := guard.GetRoomMediaUploadRegistration(ctx, roomID, registration.CommandID); readErr != nil {
|
||
return nil, readErr
|
||
} else if exists {
|
||
if !sameRoomMediaUploadIdentity(existing, registration) {
|
||
return nil, xerr.New(xerr.IdempotencyConflict, "room media upload command payload conflict")
|
||
}
|
||
switch existing.Status {
|
||
case roomMediaStatusActive:
|
||
// PutObject 已完成的同载荷重试只重放首次结果；后续权限变化不能把已成功请求改成失败。
|
||
return roomMediaUploadAuthorizationResponse(existing, s.clock.Now().UnixMilli()), nil
|
||
case "authorized":
|
||
// authorized 只代表 owner 已占用 command_id，不能代表对象已经上传成功。重试 PutObject
|
||
// 前必须重新检查当前房主/麦位/VIP，避免权限撤销后继续使用旧授权写 COS。
|
||
authorizedReplay = &existing
|
||
default:
|
||
return nil, xerr.New(xerr.Internal, "room media upload registration status is invalid")
|
||
}
|
||
}
|
||
snapshot, err := s.currentSnapshot(ctx, roomID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if snapshot == nil || snapshot.GetStatus() != state.RoomStatusActive {
|
||
return nil, xerr.New(xerr.RoomClosed, "room closed")
|
||
}
|
||
switch purpose {
|
||
case roomMediaPurposeBackground:
|
||
if snapshot.GetOwnerUserId() != actorUserID {
|
||
return nil, xerr.New(xerr.PermissionDenied, "permission denied")
|
||
}
|
||
if err := s.requireVIPBenefit(ctx, req.GetMeta().GetRequestId(), actorUserID, vipBenefitCustomRoomBackground, "upload_room_background"); err != nil {
|
||
return nil, err
|
||
}
|
||
case roomMediaPurposeImage:
|
||
if err := requireSnapshotSpeakPermission(snapshot, actorUserID, s.clock.Now().UnixMilli()); err != nil {
|
||
return nil, err
|
||
}
|
||
if err := s.requireVIPBenefit(ctx, req.GetMeta().GetRequestId(), actorUserID, vipBenefitRoomImageMessage, "upload_room_image"); err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
nowMS := s.clock.Now().UnixMilli()
|
||
if authorizedReplay != nil {
|
||
return roomMediaUploadAuthorizationResponse(*authorizedReplay, nowMS), nil
|
||
}
|
||
registration.AuthorizedVersion = snapshot.GetVersion()
|
||
registration.CreatedAtMS, registration.UpdatedAtMS = nowMS, nowMS
|
||
registered, err := guard.RegisterRoomMediaUpload(ctx, registration)
|
||
if errors.Is(err, ErrRoomMediaUploadCommandConflict) {
|
||
return nil, xerr.New(xerr.IdempotencyConflict, "room media upload command payload conflict")
|
||
}
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return roomMediaUploadAuthorizationResponse(registered, nowMS), nil
|
||
}
|
||
|
||
// CompleteRoomMediaUpload 只接受 gateway 已成功 PutObject 后回传的 URL/key。
|
||
// owner 以首次注册的完整解析事实为准推进 active，Save/Send 随后只认 active registration。
|
||
func (s *Service) CompleteRoomMediaUpload(ctx context.Context, req *roomv1.CompleteRoomMediaUploadRequest) (*roomv1.CompleteRoomMediaUploadResponse, error) {
|
||
if req == nil || req.GetMeta() == nil {
|
||
return nil, xerr.New(xerr.InvalidArgument, "request is required")
|
||
}
|
||
ctx = contextFromMeta(ctx, req.GetMeta())
|
||
roomID := strings.TrimSpace(req.GetMeta().GetRoomId())
|
||
commandID := strings.TrimSpace(req.GetMeta().GetCommandId())
|
||
actorUserID := req.GetMeta().GetActorUserId()
|
||
purpose := strings.ToLower(strings.TrimSpace(req.GetPurpose()))
|
||
if !roomid.ValidStringID(roomID) || actorUserID <= 0 || !giftlimits.ValidCommandID(commandID) ||
|
||
(purpose != roomMediaPurposeBackground && purpose != roomMediaPurposeImage) {
|
||
return nil, xerr.New(xerr.InvalidArgument, "room media completion request is invalid")
|
||
}
|
||
media := roomMediaFromProto(req.GetMedia())
|
||
if err := validateRoomMedia(appcode.FromContext(ctx), roomID, actorUserID, purpose, media); err != nil {
|
||
return nil, err
|
||
}
|
||
guard, ok := s.repository.(RoomMediaUploadGuardStore)
|
||
if !ok {
|
||
return nil, xerr.New(xerr.Unavailable, "room media upload guard is unavailable")
|
||
}
|
||
existing, exists, err := guard.GetRoomMediaUploadRegistration(ctx, roomID, commandID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if !exists {
|
||
return nil, xerr.New(xerr.NotFound, "room media upload authorization not found")
|
||
}
|
||
payloadSHA256, err := roomMediaUploadPayloadSHA256(media)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
candidate := RoomMediaUploadRegistration{
|
||
AppCode: appcode.FromContext(ctx), RoomID: roomID, CommandID: commandID, ActorUserID: actorUserID,
|
||
Purpose: purpose, MediaPayloadSHA256: payloadSHA256, ExpectedObjectKey: media.ObjectKey, ObjectURL: media.URL, Media: media,
|
||
}
|
||
if !sameRoomMediaUploadIdentity(existing, candidate) || !sameRoomMediaProperties(existing.Media, media) {
|
||
return nil, xerr.New(xerr.IdempotencyConflict, "room media upload command payload conflict")
|
||
}
|
||
if existing.Status == roomMediaStatusActive {
|
||
if strings.TrimSpace(existing.ObjectURL) != strings.TrimSpace(media.URL) {
|
||
return nil, xerr.New(xerr.IdempotencyConflict, "room media upload completion conflict")
|
||
}
|
||
return &roomv1.CompleteRoomMediaUploadResponse{
|
||
Active: true, Media: roomMediaToProto(existing.Media), ServerTimeMs: s.clock.Now().UnixMilli(),
|
||
}, nil
|
||
}
|
||
existing.ObjectURL, existing.Status, existing.Media = media.URL, roomMediaStatusActive, media
|
||
existing.UpdatedAtMS = s.clock.Now().UnixMilli()
|
||
completed, err := guard.CompleteRoomMediaUpload(ctx, existing)
|
||
if errors.Is(err, ErrRoomMediaUploadCommandConflict) {
|
||
return nil, xerr.New(xerr.IdempotencyConflict, "room media upload completion conflict")
|
||
}
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return &roomv1.CompleteRoomMediaUploadResponse{
|
||
Active: true, Media: roomMediaToProto(completed.Media), ServerTimeMs: s.clock.Now().UnixMilli(),
|
||
}, nil
|
||
}
|
||
|
||
func roomMediaUploadAuthorizationResponse(registration RoomMediaUploadRegistration, serverTimeMS int64) *roomv1.AuthorizeRoomMediaUploadResponse {
|
||
return &roomv1.AuthorizeRoomMediaUploadResponse{
|
||
Allowed: true, ObjectKeyPrefix: roomMediaObjectPrefix(registration.AppCode, registration.ActorUserID, registration.RoomID, registration.Purpose),
|
||
ObjectKey: registration.ExpectedObjectKey, RoomVersion: registration.AuthorizedVersion, ServerTimeMs: serverTimeMS,
|
||
}
|
||
}
|
||
|
||
func sameRoomMediaUploadIdentity(left RoomMediaUploadRegistration, right RoomMediaUploadRegistration) bool {
|
||
return appcode.Normalize(left.AppCode) == appcode.Normalize(right.AppCode) &&
|
||
strings.TrimSpace(left.RoomID) == strings.TrimSpace(right.RoomID) &&
|
||
strings.TrimSpace(left.CommandID) == strings.TrimSpace(right.CommandID) &&
|
||
left.ActorUserID == right.ActorUserID && strings.TrimSpace(left.Purpose) == strings.TrimSpace(right.Purpose) &&
|
||
strings.EqualFold(strings.TrimSpace(left.MediaPayloadSHA256), strings.TrimSpace(right.MediaPayloadSHA256)) &&
|
||
strings.TrimLeft(strings.TrimSpace(left.ExpectedObjectKey), "/") == strings.TrimLeft(strings.TrimSpace(right.ExpectedObjectKey), "/")
|
||
}
|
||
|
||
func (s *Service) requireActiveRoomMediaUpload(ctx context.Context, roomID string, actorUserID int64, purpose string, commandID string, media state.RoomMedia) error {
|
||
guard, ok := s.repository.(RoomMediaUploadGuardStore)
|
||
if !ok {
|
||
return xerr.New(xerr.Unavailable, "room media upload guard is unavailable")
|
||
}
|
||
registration, exists, err := guard.GetActiveRoomMediaUploadByObjectKey(ctx, media.ObjectKey)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !exists {
|
||
return xerr.New(xerr.NotFound, "active room media upload not found")
|
||
}
|
||
payloadSHA256, err := roomMediaUploadPayloadSHA256(media)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
candidate := RoomMediaUploadRegistration{
|
||
AppCode: appcode.FromContext(ctx), RoomID: roomID, CommandID: strings.TrimSpace(commandID), ActorUserID: actorUserID,
|
||
Purpose: purpose, MediaPayloadSHA256: payloadSHA256, ExpectedObjectKey: media.ObjectKey, ObjectURL: media.URL, Media: media,
|
||
}
|
||
if !sameRoomMediaUploadIdentity(registration, candidate) ||
|
||
strings.TrimSpace(registration.ObjectURL) != strings.TrimSpace(media.URL) || !sameRoomMediaProperties(registration.Media, media) {
|
||
return xerr.New(xerr.IdempotencyConflict, "room media does not match completed upload")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// SendRoomImage 把服务端认可的图片消息写入 Room Cell command log + durable outbox。
|
||
// outbox 的 at-least-once 重试使用稳定 message_id，Flutter 不能以到达次数作为消息条数。
|
||
func (s *Service) SendRoomImage(ctx context.Context, req *roomv1.SendRoomImageRequest) (*roomv1.SendRoomImageResponse, error) {
|
||
if req == nil || req.GetMeta() == nil {
|
||
return nil, xerr.New(xerr.InvalidArgument, "request is required")
|
||
}
|
||
ctx = contextFromMeta(ctx, req.GetMeta())
|
||
cmd := command.SendRoomImage{
|
||
Base: baseFromMeta(req.GetMeta()), MessageID: roomImageMessageID(req.GetMeta()), Image: roomMediaFromProto(req.GetImage()),
|
||
}
|
||
if !roomid.ValidStringID(cmd.RoomID()) || cmd.ActorUserID() <= 0 || strings.TrimSpace(cmd.ID()) == "" {
|
||
return nil, xerr.New(xerr.InvalidArgument, "room image message meta is invalid")
|
||
}
|
||
// 与装扮相同，已提交 command_id 必须在当前 VIP 到期/禁言之前直接重放。
|
||
if _, record, seen, err := s.commandRecordForIdempotency(ctx, cmd); err != nil {
|
||
return nil, err
|
||
} else if seen {
|
||
stored, deserializeErr := command.Deserialize(record.CommandType, record.Payload)
|
||
if deserializeErr != nil {
|
||
return nil, deserializeErr
|
||
}
|
||
storedCommand, ok := stored.(*command.SendRoomImage)
|
||
if !ok {
|
||
return nil, xerr.New(xerr.Internal, "stored room image command is invalid")
|
||
}
|
||
// 返回第一次提交固化的 message_id 和媒体事实，不能用当前重试请求拼出看似成功的另一条消息。
|
||
cmd = *storedCommand
|
||
} else {
|
||
if err := validateRoomMedia(appcode.FromContext(ctx), cmd.RoomID(), cmd.ActorUserID(), roomMediaPurposeImage, cmd.Image); err != nil {
|
||
return nil, err
|
||
}
|
||
if err := s.requireActiveRoomMediaUpload(ctx, cmd.RoomID(), cmd.ActorUserID(), roomMediaPurposeImage, cmd.ID(), cmd.Image); err != nil {
|
||
return nil, err
|
||
}
|
||
if err := s.requireVIPBenefit(ctx, req.GetMeta().GetRequestId(), cmd.ActorUserID(), vipBenefitRoomImageMessage, "send_room_image"); err != nil {
|
||
return nil, err
|
||
}
|
||
}
|
||
|
||
result, err := s.mutateRoom(ctx, cmd, true, func(now time.Time, current *state.RoomState, _ *rank.LocalRank) (mutationResult, []outbox.Record, error) {
|
||
if err := requireActiveRoom(current); err != nil {
|
||
return mutationResult{}, nil, err
|
||
}
|
||
if err := requireStateSpeakPermission(current, cmd.ActorUserID(), now.UnixMilli()); err != nil {
|
||
return mutationResult{}, nil, err
|
||
}
|
||
// 图片消息不成为房间核心状态字段，但版本推进把 command log、恢复谱系和 outbox 放进同一原子提交。
|
||
current.Version++
|
||
event, err := outbox.Build(current.RoomID, "RoomImageMessageSent", current.Version, now, &roomeventsv1.RoomImageMessageSent{
|
||
MessageId: cmd.MessageID, CommandId: cmd.ID(), SenderUserId: cmd.ActorUserID(), Image: roomMediaEventSnapshot(cmd.Image),
|
||
})
|
||
if err != nil {
|
||
return mutationResult{}, nil, err
|
||
}
|
||
return mutationResult{snapshot: current.ToProtoAt(now.UnixMilli())}, []outbox.Record{event}, nil
|
||
})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
now := s.clock.Now()
|
||
return &roomv1.SendRoomImageResponse{
|
||
Result: commandResult(result.applied, result.snapshot.GetVersion(), now), Accepted: true,
|
||
MessageId: cmd.MessageID, RoomId: cmd.RoomID(), SenderUserId: cmd.ActorUserID(), ServerTimeMs: now.UnixMilli(),
|
||
Image: roomMediaToProto(cmd.Image),
|
||
}, nil
|
||
}
|
||
|
||
func requireSnapshotSpeakPermission(snapshot *roomv1.RoomSnapshot, userID int64, nowMS int64) error {
|
||
if snapshot == nil || snapshot.GetStatus() != state.RoomStatusActive {
|
||
return xerr.New(xerr.RoomClosed, "room closed")
|
||
}
|
||
switch {
|
||
case snapshotUserBanned(snapshot, userID, nowMS):
|
||
return xerr.New(xerr.PermissionDenied, "user is banned")
|
||
case containsID(snapshot.GetMuteUserIds(), userID):
|
||
return xerr.New(xerr.PermissionDenied, "user is muted")
|
||
case !snapshot.GetChatEnabled():
|
||
return xerr.New(xerr.Conflict, "room chat is disabled")
|
||
case findProtoUser(snapshot, userID) == nil:
|
||
return xerr.New(xerr.PermissionDenied, "user is not in room")
|
||
default:
|
||
return nil
|
||
}
|
||
}
|
||
|
||
func requireStateSpeakPermission(current *state.RoomState, userID int64, nowMS int64) error {
|
||
if moderation, banned := current.BanUsers[userID]; banned && moderation.ActiveAt(nowMS) {
|
||
return xerr.New(xerr.PermissionDenied, "user is banned")
|
||
}
|
||
if current.MuteUsers[userID] {
|
||
return xerr.New(xerr.PermissionDenied, "user is muted")
|
||
}
|
||
if !current.ChatEnabled {
|
||
return xerr.New(xerr.Conflict, "room chat is disabled")
|
||
}
|
||
if current.OnlineUsers[userID] == nil {
|
||
return xerr.New(xerr.PermissionDenied, "user is not in room")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func roomImageMessageID(meta *roomv1.RequestMeta) string {
|
||
if meta == nil {
|
||
return ""
|
||
}
|
||
sum := sha256.Sum256([]byte(strings.Join([]string{
|
||
appcode.Normalize(meta.GetAppCode()), strings.TrimSpace(meta.GetRoomId()), strings.TrimSpace(meta.GetCommandId()),
|
||
}, "\x00")))
|
||
return "room_msg_" + hex.EncodeToString(sum[:16])
|
||
}
|