diff --git a/clients/bot/goclient/endpoints.go b/clients/bot/goclient/endpoints.go index 9110ca5..b404757 100644 --- a/clients/bot/goclient/endpoints.go +++ b/clients/bot/goclient/endpoints.go @@ -22,11 +22,10 @@ var ( EndpointGuilds = EndpointAPIPath + "/guild" EndpointGuildChannels = func(guildID string) string { return EndpointGuilds + "/" + guildID + "/channels" } - EndpointChannelMessages = func(channelID string) string { return EndpointAPIPath + "/message/channel/" + channelID } - EndpointChannelMessage = func(channelID, messageID string) string { return EndpointChannelMessages(channelID) + "/" + messageID } - EndpointChannelMessageAck = func(channelID, messageID string) string { return EndpointChannelMessage(channelID, messageID) + "/ack" } - EndpointChannelTyping = func(channelID string) string { return EndpointChannelMessages(channelID) + "/typing" } - EndpointMessageReactions = func(channelID, messageID, reaction string) string { + EndpointChannelMessages = func(channelID string) string { return EndpointAPIPath + "/message/channel/" + channelID } + EndpointChannelMessage = func(channelID, messageID string) string { return EndpointChannelMessages(channelID) + "/" + messageID } + EndpointChannelTyping = func(channelID string) string { return EndpointChannelMessages(channelID) + "/typing" } + EndpointMessageReactions = func(channelID, messageID, reaction string) string { return EndpointChannelMessage(channelID, messageID) + "/reactions/" + url.PathEscape(reaction) } EndpointMessageReactionAll = func(channelID, messageID string) string { diff --git a/clients/bot/goclient/restapi.go b/clients/bot/goclient/restapi.go index 3af1cbb..d09ef53 100644 --- a/clients/bot/goclient/restapi.go +++ b/clients/bot/goclient/restapi.go @@ -259,12 +259,6 @@ func (s *Session) ChannelMessageDelete(ctx context.Context, channelID, messageID return s.Request(ctx, http.MethodDelete, path, nil, nil, nil, options...) } -// ChannelMessageAck marks a channel read as the bot. -func (s *Session) ChannelMessageAck(ctx context.Context, channelID, messageID int64, options ...RequestOption) error { - path := EndpointChannelMessageAck(formatID(channelID), formatID(messageID)) - return s.Request(ctx, http.MethodPost, path, nil, nil, nil, options...) -} - // ChannelTyping sends a typing indicator as the bot. func (s *Session) ChannelTyping(ctx context.Context, channelID int64, options ...RequestOption) error { return s.Request(ctx, http.MethodPost, EndpointChannelTyping(formatID(channelID)), nil, nil, nil, options...) diff --git a/clients/bot/goclient/restapi_test.go b/clients/bot/goclient/restapi_test.go index f534cce..f7eac82 100644 --- a/clients/bot/goclient/restapi_test.go +++ b/clients/bot/goclient/restapi_test.go @@ -33,7 +33,6 @@ func TestBotRESTMethodsUseCurrentRuntimeRoutes(t *testing.T) { {method: http.MethodPost, path: "/bot/api/v1/message/channel/2230469276416868353", body: `{"content":"hello"}`, reply: `{"id":2230469276416868355,"channel_id":2230469276416868353,"author":{"id":1,"name":"bot","discriminator":"0001","is_bot":true},"content":"hello","type":0}`}, {method: http.MethodPatch, path: "/bot/api/v1/message/channel/2230469276416868353/2230469276416868355", body: `{"content":"edited"}`, reply: `{"id":2230469276416868355,"channel_id":2230469276416868353,"author":{"id":1,"name":"bot","discriminator":"0001","is_bot":true},"content":"edited","type":0}`}, {method: http.MethodDelete, path: "/bot/api/v1/message/channel/2230469276416868353/2230469276416868355", status: http.StatusNoContent}, - {method: http.MethodPost, path: "/bot/api/v1/message/channel/2230469276416868353/2230469276416868355/ack", status: http.StatusNoContent}, {method: http.MethodPost, path: "/bot/api/v1/message/channel/2230469276416868353/typing", status: http.StatusNoContent}, {method: http.MethodPut, path: "/bot/api/v1/message/channel/2230469276416868353/2230469276416868355/reactions/smile:123", reply: `{"count":1,"me":true,"emoji":{"id":123,"name":"smile"}}`}, {method: http.MethodDelete, path: "/bot/api/v1/message/channel/2230469276416868353/2230469276416868355/reactions/smile:123", status: http.StatusNoContent}, @@ -102,9 +101,6 @@ func TestBotRESTMethodsUseCurrentRuntimeRoutes(t *testing.T) { if err := s.ChannelMessageDelete(ctx, 2230469276416868353, 2230469276416868355); err != nil { t.Fatal(err) } - if err := s.ChannelMessageAck(ctx, 2230469276416868353, 2230469276416868355); err != nil { - t.Fatal(err) - } if err := s.ChannelTyping(ctx, 2230469276416868353); err != nil { t.Fatal(err) } diff --git a/cmd/botapi/endpoints/applicationcommand/handlers.go b/cmd/botapi/endpoints/applicationcommand/handlers.go index f6ad074..5dec60e 100644 --- a/cmd/botapi/endpoints/applicationcommand/handlers.go +++ b/cmd/botapi/endpoints/applicationcommand/handlers.go @@ -1,10 +1,9 @@ package applicationcommand import ( - "time" - appcmd "github.com/FlameInTheDark/gochat/internal/applicationcommands" "github.com/FlameInTheDark/gochat/internal/botauth" + "github.com/FlameInTheDark/gochat/internal/dto" "github.com/FlameInTheDark/gochat/internal/idgen" "github.com/gofiber/fiber/v2" ) @@ -190,9 +189,6 @@ func (e *Entity) RespondInteraction(c *fiber.Ctx) error { if err := requireInteractionID(c, record); err != nil { return err } - if time.Now().After(record.CreatedAt.Add(appcmd.InteractionDeadline)) { - return fiber.NewError(fiber.StatusNotFound, "unknown interaction") - } var response appcmd.InteractionResponse if err := c.BodyParser(&response); err != nil { return fiber.NewError(fiber.StatusBadRequest, "invalid interaction response") @@ -202,21 +198,30 @@ func (e *Entity) RespondInteraction(c *fiber.Ctx) error { if err := e.appcmd.AckInteraction(c.UserContext(), record.ID, appcmd.AckStateResponded, nil); err != nil { return duplicateAckError() } + e.notifyInteractionStatus(c.UserContext(), record, "responded", response.Data, nil) return c.SendStatus(fiber.StatusNoContent) case appcmd.ResponseTypeDeferredChannelMessageSource: + var deferredData *appcmd.InteractionResponseData if response.Data != nil && appcmd.HasResponseFlag(response.Data, appcmd.MessageFlagEphemeral) { - _ = e.setEphemeralResponse(c.UserContext(), record.ID, &appcmd.InteractionResponseData{Flags: response.Data.Flags | appcmd.MessageFlagLoading}) + deferredData = &appcmd.InteractionResponseData{Flags: response.Data.Flags | appcmd.MessageFlagLoading} + _ = e.setEphemeralResponse(c.UserContext(), record.ID, deferredData) + } else { + deferredData = response.Data } if err := e.appcmd.AckInteraction(c.UserContext(), record.ID, appcmd.AckStateDeferred, nil); err != nil { return duplicateAckError() } + e.notifyInteractionStatus(c.UserContext(), record, "deferred", deferredData, nil) return c.SendStatus(fiber.StatusNoContent) case appcmd.ResponseTypeChannelMessageWithSource: var messageID *int64 + var message *dto.Message + state := "responded" if appcmd.HasResponseFlag(response.Data, appcmd.MessageFlagEphemeral) { if err := e.setEphemeralResponse(c.UserContext(), record.ID, response.Data); err != nil { return fiber.NewError(fiber.StatusInternalServerError, "unable to store ephemeral response") } + state = "ephemeral" } else { msg, err := e.createPublicInteractionMessage(c, record, response.Data) if err != nil { @@ -224,10 +229,12 @@ func (e *Entity) RespondInteraction(c *fiber.Ctx) error { } id := msg.Id messageID = &id + message = &msg } if err := e.appcmd.AckInteraction(c.UserContext(), record.ID, appcmd.AckStateResponded, messageID); err != nil { return duplicateAckError() } + e.notifyInteractionStatus(c.UserContext(), record, state, response.Data, message) return c.SendStatus(fiber.StatusNoContent) case appcmd.ResponseTypeAutocompleteResult: if response.Data == nil { @@ -247,6 +254,7 @@ func (e *Entity) RespondInteraction(c *fiber.Ctx) error { if err := e.appcmd.AckInteraction(c.UserContext(), record.ID, appcmd.AckStateResponded, nil); err != nil { return duplicateAckError() } + e.notifyInteractionStatus(c.UserContext(), record, "modal", response.Data, nil) return c.JSON(response) default: return fiber.NewError(fiber.StatusBadRequest, "unsupported interaction response type") @@ -294,6 +302,7 @@ func (e *Entity) EditOriginalInteractionResponse(c *fiber.Ctx) error { if err := e.setEphemeralResponse(c.UserContext(), record.ID, &data); err != nil { return fiber.NewError(fiber.StatusInternalServerError, "unable to store ephemeral response") } + e.notifyInteractionStatus(c.UserContext(), record, "ephemeral", &data, nil) return c.JSON(data) } if record.InitialResponseID == nil { @@ -303,6 +312,7 @@ func (e *Entity) EditOriginalInteractionResponse(c *fiber.Ctx) error { if err := e.setEphemeralResponse(c.UserContext(), record.ID, &data); err != nil { return fiber.NewError(fiber.StatusInternalServerError, "unable to store ephemeral response") } + e.notifyInteractionStatus(c.UserContext(), record, "ephemeral", &data, nil) return c.JSON(data) } out, err := e.createPublicInteractionMessage(c, record, &data) @@ -313,12 +323,14 @@ func (e *Entity) EditOriginalInteractionResponse(c *fiber.Ctx) error { _ = e.deleteMessageResponse(c, record, out.Id) return duplicateAckError() } + e.notifyInteractionStatus(c.UserContext(), record, "responded", &data, &out) return c.JSON(out) } out, err := e.editMessageResponse(c, record, *record.InitialResponseID, &data) if err != nil { return err } + e.notifyInteractionStatus(c.UserContext(), record, "responded", &data, &out) return c.JSON(out) } @@ -357,12 +369,14 @@ func (e *Entity) CreateFollowupMessage(c *fiber.Ctx) error { if err := e.setEphemeralResponse(c.UserContext(), idgen.Next(), &data); err != nil { return fiber.NewError(fiber.StatusInternalServerError, "unable to store ephemeral followup") } + e.notifyInteractionStatus(c.UserContext(), record, "ephemeral_followup", &data, nil) return c.Status(fiber.StatusCreated).JSON(data) } out, err := e.createPublicInteractionMessage(c, record, &data) if err != nil { return err } + e.notifyInteractionStatus(c.UserContext(), record, "responded", &data, &out) return c.Status(fiber.StatusCreated).JSON(out) } diff --git a/cmd/botapi/endpoints/applicationcommand/helpers.go b/cmd/botapi/endpoints/applicationcommand/helpers.go index f6b8de5..11fff8f 100644 --- a/cmd/botapi/endpoints/applicationcommand/helpers.go +++ b/cmd/botapi/endpoints/applicationcommand/helpers.go @@ -239,6 +239,27 @@ func (e *Entity) deleteEphemeralResponse(ctx context.Context, interactionID int6 } } +func (e *Entity) notifyInteractionStatus(ctx context.Context, record appcmd.InteractionRecord, state string, data *appcmd.InteractionResponseData, message *dto.Message) { + commandName := "" + if payload, err := e.payload.GetInteractionPayload(ctx, record.ID); err == nil { + if parsed := appcmd.ParseCommandData(payload.DataJSON); parsed != nil { + commandName = parsed.Name + } + } + _ = mq.SendUserUpdate(ctx, e.mqt, record.InvokerUserID, &mqmsg.ApplicationCommandInteractionStatus{ + InteractionID: record.ID, + ApplicationID: record.ApplicationID, + CommandID: record.CommandID, + CommandName: commandName, + ChannelID: record.ChannelID, + GuildID: record.GuildID, + UserID: record.InvokerUserID, + State: state, + Response: data, + Message: message, + }) +} + func (e *Entity) setAutocompleteResponse(ctx context.Context, interactionID int64, choices []appcmd.ApplicationCommandChoice) error { if e.cache == nil { return nil diff --git a/cmd/botapi/endpoints/message/entity.go b/cmd/botapi/endpoints/message/entity.go index ebb759f..1e17bcf 100644 --- a/cmd/botapi/endpoints/message/entity.go +++ b/cmd/botapi/endpoints/message/entity.go @@ -63,7 +63,6 @@ func (e *Entity) Init(router fiber.Router) { router.Get("/channel/:channel_id", e.List) router.Patch("/channel/:channel_id/:message_id", e.Update) router.Delete("/channel/:channel_id/:message_id", e.Delete) - router.Post("/channel/:channel_id/:message_id/ack", e.Ack) router.Post("/channel/:channel_id/typing", e.Typing) router.Put("/channel/:channel_id/:message_id/reactions/:reaction_name", e.AddReaction) router.Delete("/channel/:channel_id/:message_id/reactions/:reaction_name", e.RemoveReaction) diff --git a/cmd/botapi/endpoints/message/handlers.go b/cmd/botapi/endpoints/message/handlers.go index 7bd9ab0..4b84901 100644 --- a/cmd/botapi/endpoints/message/handlers.go +++ b/cmd/botapi/endpoints/message/handlers.go @@ -276,36 +276,6 @@ func (e *Entity) Delete(c *fiber.Ctx) error { return c.SendStatus(fiber.StatusNoContent) } -// Ack -// -// @Summary Mark a channel read as the bot -// @Produce json -// @Tags Bot Message -// @Security BotToken -// @Param channel_id path int64 true "Channel id" -// @Param message_id path int64 true "Message id" -// @Success 204 -// @Failure 400 {string} string "Bad request" -// @Failure 401 {string} string "Unauthorized" -// @Failure 403 {string} string "Forbidden" -// @Failure 500 {string} string "Internal server error" -// @Router /bot/api/v1/message/channel/{channel_id}/{message_id}/ack [post] -func (e *Entity) Ack(c *fiber.Ctx) error { - principal, channel, _, err := e.requireChannel(c, permissions.PermServerViewChannels, permissions.PermTextReadMessageHistory) - if err != nil { - return err - } - msgID, err := parseParamInt64(c, "message_id") - if err != nil { - return err - } - if err := e.rs.SetReadState(c.UserContext(), principal.BotUserID, channel.Id, msgID); err != nil { - return fiber.NewError(fiber.StatusInternalServerError, "unable to update read state") - } - _ = mq.SendUserUpdate(c.UserContext(), e.mqt, principal.BotUserID, &mqmsg.UpdateReadState{ChannelId: channel.Id, MessageId: msgID}) - return c.SendStatus(fiber.StatusNoContent) -} - // Typing // // @Summary Send a typing indicator as the bot diff --git a/cmd/ws/app.go b/cmd/ws/app.go index 9925e01..2fb8b88 100644 --- a/cmd/ws/app.go +++ b/cmd/ws/app.go @@ -181,7 +181,7 @@ func (a *App) presenceTTLSeconds() int64 { if a == nil || a.cfg == nil { return 60 } - ttl := a.cfg.HearthBeatTimeout * 2 / 1000 + ttl := int64((wsHeartbeatDeadline(a.cfg.HearthBeatTimeout) * 2) / time.Second) if ttl < 1 { return 1 } diff --git a/cmd/ws/handler.go b/cmd/ws/handler.go index 195d3cc..855cea5 100644 --- a/cmd/ws/handler.go +++ b/cmd/ws/handler.go @@ -276,10 +276,7 @@ func (a *App) wsHandler(c *websocket.Conn) { pingInterval = half } } - readDeadline := time.Duration(a.cfg.HearthBeatTimeout+15000) * time.Millisecond - if readDeadline <= 0 { - readDeadline = 60 * time.Second - } + readDeadline := wsReadDeadline(a.cfg.HearthBeatTimeout) _ = c.SetReadDeadline(time.Now().Add(readDeadline)) c.SetPongHandler(func(string) error { return c.SetReadDeadline(time.Now().Add(readDeadline)) @@ -317,6 +314,7 @@ func (a *App) wsHandler(c *websocket.Conn) { connLog.Error("Read WS message error", "error", err) return } + _ = c.SetReadDeadline(time.Now().Add(readDeadline)) switch mt { case websocket.TextMessage: @@ -356,3 +354,22 @@ func isExpectedWSReadError(err error) bool { return errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) || errors.Is(err, net.ErrClosed) } + +func wsReadDeadline(heartbeatTimeoutMs int64) time.Duration { + return wsHeartbeatDeadline(heartbeatTimeoutMs) + 15*time.Second +} + +func wsHeartbeatDeadline(heartbeatTimeoutMs int64) time.Duration { + base := time.Duration(heartbeatTimeoutMs) * time.Millisecond + if base <= 0 { + return 60 * time.Second + } + deadline := base + 10*time.Second + if minimum := base * 3; deadline < minimum { + deadline = minimum + } + if deadline < 60*time.Second { + return 60 * time.Second + } + return deadline +} diff --git a/cmd/ws/handler/handler.go b/cmd/ws/handler/handler.go index 1df6a8a..500c81f 100644 --- a/cmd/ws/handler/handler.go +++ b/cmd/ws/handler/handler.go @@ -8,6 +8,7 @@ import ( "log/slog" "reflect" "sort" + "sync/atomic" "time" "github.com/FlameInTheDark/gochat/internal/database/pgentities/rolecheck" @@ -103,6 +104,7 @@ type Handler struct { generation int64 lastEventId int64 hbTimeout int64 + closed atomic.Bool // Whether we successfully set presence after hello presenceSet bool } @@ -168,36 +170,36 @@ func (h *Handler) HandleMessage(e mqmsg.Message) { log.Warn("Error unmarshalling heart beat msg", "error", err) return } - if m.LastEventId >= h.lastEventId { + if h.hTimer != nil { + h.hTimer.Reset(h.heartbeatDeadline()) + } + accepted := m.LastEventId >= h.lastEventId + if accepted { h.telemetry.Heartbeat(ctx, "accepted") - // add grace to tolerate network jitter (10s) - h.hTimer.Reset(time.Millisecond * time.Duration(h.hbTimeout+10000)) - if h.gstate != nil && h.connectionID != "" { - opCtx, cancel := context.WithTimeout(ctx, time.Second) - ttl := h.connectionTTLSeconds() - _ = h.gstate.TouchConnection(opCtx, h.connectionID, m.LastEventId, ttl) - cancel() - } - // Refresh this session TTL: heartbeat_interval * 2 - // Throttled: skip if we touched within the last 10s. - if h.user != nil && h.pstore != nil && h.sessionID != "" && h.presenceSet && - time.Since(h.lastPresenceTouch) > 10*time.Second { - opCtx, cancel := context.WithTimeout(ctx, time.Second*2) - // TTL expects seconds - ttl := h.presenceTTLSeconds() - if h.connectionID != "" && h.generation > 0 { - _ = h.pstore.TouchSessionTTLIfOwner(opCtx, h.user.Id, h.sessionID, h.connectionID, h.generation, ttl) - } else { - _ = h.pstore.TouchSessionTTL(opCtx, h.user.Id, h.sessionID, ttl) - } - cancel() - h.lastPresenceTouch = time.Now() - } h.lastEventId = m.LastEventId - _ = h.sendHeartbeatAck(m.LastEventId) } else { h.telemetry.Heartbeat(ctx, "stale") } + if h.gstate != nil && h.connectionID != "" { + opCtx, cancel := context.WithTimeout(ctx, time.Second) + ttl := h.connectionTTLSeconds() + _ = h.gstate.TouchConnection(opCtx, h.connectionID, h.lastEventId, ttl) + cancel() + } + // Refresh this session TTL. Throttled: skip if we touched within the last 10s. + if h.user != nil && h.pstore != nil && h.sessionID != "" && h.presenceSet && + time.Since(h.lastPresenceTouch) > 10*time.Second { + opCtx, cancel := context.WithTimeout(ctx, time.Second*2) + ttl := h.presenceTTLSeconds() + if h.connectionID != "" && h.generation > 0 { + _ = h.pstore.TouchSessionTTLIfOwner(opCtx, h.user.Id, h.sessionID, h.connectionID, h.generation, ttl) + } else { + _ = h.pstore.TouchSessionTTL(opCtx, h.user.Id, h.sessionID, ttl) + } + cancel() + h.lastPresenceTouch = time.Now() + } + _ = h.sendHeartbeatAck(m.LastEventId) case mqmsg.OPCodeChannelSubscription: var m mqmsg.Subscribe err := json.Unmarshal(e.Data, &m) @@ -323,7 +325,7 @@ func (h *Handler) HandleMessage(e mqmsg.Message) { } agg, _, _ := h.pstore.Aggregate(opCtx, h.user.Id, time.Now().Unix()) // cache aggregated presence and publish - _ = h.pstore.SetAggregated(opCtx, agg, h.hbTimeout*2/1000) + _ = h.pstore.SetAggregated(opCtx, agg, h.presenceTTLSeconds()) h.publishPresence(agg) } cancel() @@ -592,6 +594,15 @@ func (h *Handler) canSubscribeChannel(ctx context.Context, channelID int64) bool } func (h *Handler) Close() error { + if h.closed.Swap(true) { + return nil + } + if h.initTimer != nil { + h.initTimer.Stop() + } + if h.hTimer != nil { + h.hTimer.Stop() + } h.OnWSClosed() h.closer() return nil @@ -691,7 +702,7 @@ func (h *Handler) publishPresence(agg presence.Presence) { } func (h *Handler) presenceTTLSeconds() int64 { - ttl := h.hbTimeout * 2 / 1000 + ttl := int64((h.heartbeatDeadline() * 2) / time.Second) if ttl < 1 { return 1 } @@ -699,13 +710,28 @@ func (h *Handler) presenceTTLSeconds() int64 { } func (h *Handler) connectionTTLSeconds() int64 { - ttl := (h.hbTimeout + 10000) * 2 / 1000 + ttl := int64((h.heartbeatDeadline() * 2) / time.Second) if ttl < 1 { return 1 } return ttl } +func (h *Handler) heartbeatDeadline() time.Duration { + base := time.Duration(h.hbTimeout) * time.Millisecond + if base <= 0 { + return 60 * time.Second + } + deadline := base + 10*time.Second + if minimum := base * 3; deadline < minimum { + deadline = minimum + } + if deadline < 60*time.Second { + return 60 * time.Second + } + return deadline +} + func (h *Handler) clientStateTTLSeconds() int64 { return int64((24 * time.Hour) / time.Second) } diff --git a/cmd/ws/handler/heartbeat_test.go b/cmd/ws/handler/heartbeat_test.go new file mode 100644 index 0000000..c82b982 --- /dev/null +++ b/cmd/ws/handler/heartbeat_test.go @@ -0,0 +1,60 @@ +package handler + +import ( + "encoding/json" + "testing" + "time" + + "github.com/FlameInTheDark/gochat/internal/dto" + "github.com/FlameInTheDark/gochat/internal/mq/mqmsg" +) + +func TestHeartbeatDeadlineToleratesBrowserTimerThrottling(t *testing.T) { + t.Parallel() + + h := &Handler{hbTimeout: 35_000} + if got, want := h.heartbeatDeadline(), 105*time.Second; got != want { + t.Fatalf("heartbeatDeadline() = %s, want %s", got, want) + } +} + +func TestStaleHeartbeatStillGetsAcked(t *testing.T) { + t.Parallel() + + var sent mqmsg.Message + h := &Handler{ + user: &dto.User{Id: 1}, + lastEventId: 10, + hbTimeout: 35_000, + hTimer: time.NewTimer(time.Hour), + sendJSON: func(v any) error { + var ok bool + sent, ok = v.(mqmsg.Message) + if !ok { + t.Fatalf("sent payload type = %T, want mqmsg.Message", v) + } + return nil + }, + } + defer h.hTimer.Stop() + + data, err := json.Marshal(heartbeatMessage{LastEventId: 5}) + if err != nil { + t.Fatal(err) + } + h.HandleMessage(mqmsg.Message{Operation: mqmsg.OPCodeHeartBeat, Data: data}) + + if h.lastEventId != 10 { + t.Fatalf("lastEventId regressed to %d", h.lastEventId) + } + if sent.Operation != mqmsg.OPCodeHeartbeatAck { + t.Fatalf("sent op = %d, want %d", sent.Operation, mqmsg.OPCodeHeartbeatAck) + } + var ack mqmsg.HeartbeatAck + if err := json.Unmarshal(sent.Data, &ack); err != nil { + t.Fatal(err) + } + if ack.LastEventID != 5 { + t.Fatalf("ack e = %d, want 5", ack.LastEventID) + } +} diff --git a/cmd/ws/handler/hello.go b/cmd/ws/handler/hello.go index f0a91dc..f5754e6 100644 --- a/cmd/ws/handler/hello.go +++ b/cmd/ws/handler/hello.go @@ -84,7 +84,7 @@ func (h *Handler) hello(msg *mqmsg.Message) { } // --- Parallel DB fetch: user + guilds --- - ctx, cancel := context.WithTimeout(helper.ContextWithUserID(ctx, token.UserID), time.Second*time.Duration(h.hbTimeout)) + ctx, cancel := context.WithTimeout(helper.ContextWithUserID(ctx, token.UserID), time.Millisecond*time.Duration(h.hbTimeout)) defer cancel() type userResult struct { @@ -191,7 +191,10 @@ func (h *Handler) hello(msg *mqmsg.Message) { log.Error("Error sending hello message", "error", err) return } - h.hTimer = time.AfterFunc(time.Millisecond*time.Duration(h.hbTimeout+10000), func() { + h.hTimer = time.AfterFunc(h.heartbeatDeadline(), func() { + if h.closed.Load() { + return + } timeoutCtx := h.baseContext() h.telemetry.HeartbeatTimeout(timeoutCtx) helper.WithContext(h.log, timeoutCtx).Warn("Heartbeat timeout; closing WS", "user_id", func() any { diff --git a/cmd/ws/handler_read_error_test.go b/cmd/ws/handler_read_error_test.go index 490e585..a7c4f40 100644 --- a/cmd/ws/handler_read_error_test.go +++ b/cmd/ws/handler_read_error_test.go @@ -5,6 +5,7 @@ import ( "io" "net" "testing" + "time" fws "github.com/fasthttp/websocket" ) @@ -55,3 +56,11 @@ func TestIsExpectedWSReadError(t *testing.T) { }) } } + +func TestWSReadDeadlineUsesHeartbeatGrace(t *testing.T) { + t.Parallel() + + if got, want := wsReadDeadline(35_000), 120*time.Second; got != want { + t.Fatalf("wsReadDeadline() = %s, want %s", got, want) + } +} diff --git a/internal/mq/mqmsg/application_command.go b/internal/mq/mqmsg/application_command.go index 925de3c..fa2371b 100644 --- a/internal/mq/mqmsg/application_command.go +++ b/internal/mq/mqmsg/application_command.go @@ -4,6 +4,7 @@ import ( "encoding/json" appcmd "github.com/FlameInTheDark/gochat/internal/applicationcommands" + "github.com/FlameInTheDark/gochat/internal/dto" ) type ApplicationCommandInteractionCreate struct { @@ -22,3 +23,29 @@ func (m *ApplicationCommandInteractionCreate) Operation() OPCodeType { func (m *ApplicationCommandInteractionCreate) Marshal() ([]byte, error) { return json.Marshal(m) } + +type ApplicationCommandInteractionStatus struct { + InteractionID int64 `json:"interaction_id"` + ApplicationID int64 `json:"application_id"` + CommandID int64 `json:"command_id"` + CommandName string `json:"command_name,omitempty"` + ChannelID int64 `json:"channel_id"` + GuildID *int64 `json:"guild_id,omitempty"` + UserID int64 `json:"user_id"` + State string `json:"state"` + Response *appcmd.InteractionResponseData `json:"response,omitempty"` + Message *dto.Message `json:"message,omitempty"` +} + +func (m *ApplicationCommandInteractionStatus) EventType() *EventType { + t := EventTypeApplicationCommandInteractionStatus + return &t +} + +func (m *ApplicationCommandInteractionStatus) Operation() OPCodeType { + return OpCodeDispatch +} + +func (m *ApplicationCommandInteractionStatus) Marshal() ([]byte, error) { + return json.Marshal(m) +} diff --git a/internal/mq/mqmsg/base.go b/internal/mq/mqmsg/base.go index d4b43db..3717cbe 100644 --- a/internal/mq/mqmsg/base.go +++ b/internal/mq/mqmsg/base.go @@ -151,6 +151,7 @@ const ( const ( EventTypeApplicationCommandInteractionCreate EventType = 600 + iota + EventTypeApplicationCommandInteractionStatus ) type Message struct {