26129195806

Committed 19 May 2026 10:29PM UTC coverage: 76.957% (+0.05%) from 76.903%

Build # 26129195806

Build Type

push

github

Committed by

et-nik

Commit Message

remove legacy

Coverage Stats

16 of 29 new or added lines in 6 files covered. (55.17%)

574 existing lines in 7 files now uncovered.

42926 of 55779 relevant lines covered (76.96%)

35881.19 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

82.08

/internal/api/ws/attach/handler.go

package attach

import (
        "context"
        "encoding/json"
        "log/slog"
        "net/http"

        "github.com/coder/websocket"
        "github.com/gameap/gameap/internal/api/base"
        serversbase "github.com/gameap/gameap/internal/api/servers/base"
        "github.com/gameap/gameap/internal/domain"
        "github.com/gameap/gameap/internal/filters"
        "github.com/gameap/gameap/internal/grpc/handlers"
        "github.com/gameap/gameap/internal/grpc/session"
        "github.com/gameap/gameap/internal/pubsub/channels"
        "github.com/gameap/gameap/internal/repositories"
        "github.com/gameap/gameap/internal/ws"
        "github.com/gameap/gameap/pkg/api"
        "github.com/gameap/gameap/pkg/auth"
        "github.com/gameap/gameap/pkg/idgen"
        "github.com/gameap/gameap/pkg/proto"
        "github.com/pkg/errors"
)

const (
        typeAttachInput  = "attach.input"
        typeAttachDetach = "attach.detach"
)

type Handler struct {
        serverFinder   *serversbase.ServerFinder
        abilityChecker *serversbase.AbilityChecker
        nodeRepo       repositories.NodeRepository
        hub            *ws.Hub
        originPatterns []string
        registry       *session.Registry
        attachHandler  *handlers.AttachHandler
        responder      base.Responder
        logger         *slog.Logger
}

func NewHandler(
        serverRepo repositories.ServerRepository,
        nodeRepo repositories.NodeRepository,
        rbac base.RBAC,
        hub *ws.Hub,
        originPatterns []string,
        registry *session.Registry,
        attachHandler *handlers.AttachHandler,
        responder base.Responder,
) *Handler {
        return &Handler{
                serverFinder:   serversbase.NewServerFinder(serverRepo, rbac),
                abilityChecker: serversbase.NewAbilityChecker(rbac),
                nodeRepo:       nodeRepo,
                hub:            hub,
                originPatterns: originPatterns,
                registry:       registry,
                attachHandler:  attachHandler,
                responder:      responder,
                logger:         slog.Default(),
        }
}

func (h *Handler) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
        ctx := r.Context()

        sess := auth.SessionFromContext(ctx)
        if !sess.IsAuthenticated() {
                h.responder.WriteError(ctx, rw, api.NewError(http.StatusUnauthorized, "user not authenticated"))

                return
        }

        input := api.NewInputReader(r)

        serverID, err := input.ReadUint("server")
        if err != nil {
                h.responder.WriteError(ctx, rw, api.WrapHTTPError(
                        errors.WithMessage(err, "invalid server id"),
                        http.StatusBadRequest,
                ))

                return
        }

        server, err := h.serverFinder.FindUserServer(ctx, sess.User, serverID)
        if err != nil {
                h.responder.WriteError(ctx, rw, err)

                return
        }

        if err = h.abilityChecker.CheckOrError(
                ctx,
                sess.User.ID,
                server.ID,
                []domain.AbilityName{domain.AbilityNameGameServerConsoleView},
        ); err != nil {
                h.responder.WriteError(ctx, rw, err)

                return
        }

        node, err := h.findNode(ctx, server.DSID)
        if err != nil {
                h.responder.WriteError(ctx, rw, err)

                return
        }

        nodeID := uint64(server.DSID)

        if !h.registry.IsConnectedAnywhere(nodeID) {
                h.responder.WriteError(ctx, rw, api.NewError(
                        http.StatusServiceUnavailable,
                        "daemon is not connected via grpc",
                ))

                return
        }

        conn, err := ws.Accept(rw, r, &websocket.AcceptOptions{
                OriginPatterns: h.originPatterns,
        })
        if err != nil {
                h.logger.Warn("websocket accept failed", "error", err)

                return
        }

        canSend := h.canSendCommands(ctx, sess.User, server)

        sessionID := idgen.New()
        h.runAttachSession(ctx, conn, server, node, sessionID, sess.User, canSend)
}

func (h *Handler) runAttachSession(
        ctx context.Context,
        conn *websocket.Conn,
        server *domain.Server,
        node *domain.Node,
        sessionID string,
        user *domain.User,
        canSend bool,
) {
        client := ws.NewClient(ctx, conn, h.hub, nil, h.logger)

        startedTopic := ws.ChannelToTopic(channels.BuildRealtimeAttachStartedChannel(sessionID))
        outputTopic := ws.ChannelToTopic(channels.BuildRealtimeAttachOutputChannel(sessionID))
        closedTopic := ws.ChannelToTopic(channels.BuildRealtimeAttachClosedChannel(sessionID))

        h.hub.Register(client, startedTopic, outputTopic, closedTopic)
        h.attachHandler.TrackAttachSession(sessionID, uint64(server.ID))

        nodeID := uint64(node.ID)

        msgHandler := h.newMessageHandler(ctx, client, server, node, user, sessionID, canSend)
        client.SetMessageHandler(msgHandler)

        if err := h.registry.SendAttachRequest(ctx, nodeID, &proto.AttachRequest{
                SessionId: sessionID,
                ServerId:  uint64(server.ID),
        }); err != nil {
                h.logger.Warn("failed to send attach request",
                        "server_id", server.ID,
                        "session_id", sessionID,
                        "error", err,
                )
                client.SendMessage(ws.NewErrorMessage("failed to attach to server console"))
        }

        client.Run()

        _ = h.registry.SendAttachDetach(context.Background(), nodeID, &proto.AttachDetach{
                SessionId: sessionID,
                Reason:    "client disconnected",
        })
        h.attachHandler.UntrackAttachSession(sessionID)
}

func (h *Handler) newMessageHandler(
        ctx context.Context,
        client *ws.Client,
        server *domain.Server,
        node *domain.Node,
        user *domain.User,
        sessionID string,
        canSend bool,
) ws.MessageHandler {
        nodeID := uint64(node.ID)

        return func(_ context.Context, msg *ws.InboundMessage) {
                switch msg.Type {
                case typeAttachInput:
                        if !canSend {
                                client.SendMessage(ws.NewErrorMessage("permission denied: cannot send input"))

                                return
                        }

                        if err := h.abilityChecker.CheckOrError(
                                ctx,
                                user.ID,
                                server.ID,
                                []domain.AbilityName{domain.AbilityNameGameServerConsoleSend},
                        ); err != nil {
                                client.SendMessage(ws.NewErrorMessage("permission denied: cannot send input"))

                                return
                        }

                        var payload attachInputPayload
                        if err := json.Unmarshal(msg.Payload, &payload); err != nil {
                                return
                        }

                        if err := h.registry.SendAttachInput(ctx, nodeID, &proto.AttachInput{
                                SessionId: sessionID,
                                Data:      payload.Data,
                        }); err != nil {
                                h.logger.Warn("failed to send attach input",
                                        "session_id", sessionID,
                                        "error", err,
                                )
                        }

                case typeAttachDetach:
                        _ = h.registry.SendAttachDetach(ctx, nodeID, &proto.AttachDetach{
                                SessionId: sessionID,
                                Reason:    "user detached",
                        })
                        client.Close()
                }
        }
}

func (h *Handler) findNode(ctx context.Context, nodeID uint) (*domain.Node, error) {
        nodes, err := h.nodeRepo.Find(ctx, &filters.FindNode{
                IDs: []uint{nodeID},
        }, nil, &filters.Pagination{
                Limit: 1,
        })
        if err != nil {
                return nil, errors.WithMessage(err, "failed to find node")
        }

        if len(nodes) == 0 {
                return nil, api.NewNotFoundError("node not found")
        }

        return &nodes[0], nil
}

func (h *Handler) canSendCommands(ctx context.Context, user *domain.User, server *domain.Server) bool {
        err := h.abilityChecker.CheckOrError(
                ctx,
                user.ID,
                server.ID,
                []domain.AbilityName{domain.AbilityNameGameServerConsoleSend},
        )

        return err == nil
}

type attachInputPayload struct {
        Data []byte `json:"data"`
}

1	package attach
2
3	import (
4	"context"
5	"encoding/json"
6	"log/slog"
7	"net/http"
8
9	"github.com/coder/websocket"
10	"github.com/gameap/gameap/internal/api/base"
11	serversbase "github.com/gameap/gameap/internal/api/servers/base"
12	"github.com/gameap/gameap/internal/domain"
13	"github.com/gameap/gameap/internal/filters"
14	"github.com/gameap/gameap/internal/grpc/handlers"
15	"github.com/gameap/gameap/internal/grpc/session"
16	"github.com/gameap/gameap/internal/pubsub/channels"
17	"github.com/gameap/gameap/internal/repositories"
18	"github.com/gameap/gameap/internal/ws"
19	"github.com/gameap/gameap/pkg/api"
20	"github.com/gameap/gameap/pkg/auth"
21	"github.com/gameap/gameap/pkg/idgen"
22	"github.com/gameap/gameap/pkg/proto"
23	"github.com/pkg/errors"
24	)
25
26	const (
27	typeAttachInput = "attach.input"
28	typeAttachDetach = "attach.detach"
29	)
30
31	type Handler struct {
32	serverFinder *serversbase.ServerFinder
33	abilityChecker *serversbase.AbilityChecker
34	nodeRepo repositories.NodeRepository
35	hub *ws.Hub
36	originPatterns []string
37	registry *session.Registry
38	attachHandler *handlers.AttachHandler
39	responder base.Responder
40	logger *slog.Logger
41	}
42
43	func NewHandler(
44	serverRepo repositories.ServerRepository,
45	nodeRepo repositories.NodeRepository,
46	rbac base.RBAC,
47	hub *ws.Hub,
48	originPatterns []string,
49	registry *session.Registry,
50	attachHandler *handlers.AttachHandler,
51	responder base.Responder,
52	) *Handler {	16✔
53	return &Handler{	16✔
54	serverFinder: serversbase.NewServerFinder(serverRepo, rbac),	16✔
55	abilityChecker: serversbase.NewAbilityChecker(rbac),	16✔
56	nodeRepo: nodeRepo,	16✔
57	hub: hub,	16✔
58	originPatterns: originPatterns,	16✔
59	registry: registry,	16✔
60	attachHandler: attachHandler,	16✔
61	responder: responder,	16✔
62	logger: slog.Default(),	16✔
63	}	16✔
64	}	16✔
65
66	func (h Handler) ServeHTTP(rw http.ResponseWriter, r http.Request) {	11✔
67	ctx := r.Context()	11✔
68		11✔
69	sess := auth.SessionFromContext(ctx)	11✔
70	if !sess.IsAuthenticated() {	12✔
71	h.responder.WriteError(ctx, rw, api.NewError(http.StatusUnauthorized, "user not authenticated"))	1✔
72		1✔
73	return	1✔
74	}	1✔
75
76	input := api.NewInputReader(r)	10✔
77		10✔
78	serverID, err := input.ReadUint("server")	10✔
79	if err != nil {	11✔
80	h.responder.WriteError(ctx, rw, api.WrapHTTPError(	1✔
81	errors.WithMessage(err, "invalid server id"),	1✔
82	http.StatusBadRequest,	1✔
83	))	1✔
84		1✔
85	return	1✔
86	}	1✔
87
88	server, err := h.serverFinder.FindUserServer(ctx, sess.User, serverID)	9✔
89	if err != nil {	10✔
90	h.responder.WriteError(ctx, rw, err)	1✔
91		1✔
92	return	1✔
93	}	1✔
94
95	if err = h.abilityChecker.CheckOrError(	8✔
96	ctx,	8✔
97	sess.User.ID,	8✔
98	server.ID,	8✔
99	[]domain.AbilityName{domain.AbilityNameGameServerConsoleView},	8✔
100	); err != nil {	9✔
101	h.responder.WriteError(ctx, rw, err)	1✔
102		1✔
103	return	1✔
104	}	1✔
105
106	node, err := h.findNode(ctx, server.DSID)	7✔
107	if err != nil {	8✔
108	h.responder.WriteError(ctx, rw, err)	1✔
109		1✔
110	return	1✔
111	}	1✔
112
113	nodeID := uint64(server.DSID)	6✔
114		6✔
115	if !h.registry.IsConnectedAnywhere(nodeID) {	6✔
NEW 116	h.responder.WriteError(ctx, rw, api.NewError(	×
NEW 117	http.StatusServiceUnavailable,	×
NEW 118	"daemon is not connected via grpc",	×
NEW 119	))	×
NEW 120		×
NEW 121	return	×
NEW 122	}	×
123
124	conn, err := ws.Accept(rw, r, &websocket.AcceptOptions{	6✔
125	OriginPatterns: h.originPatterns,	6✔
126	})	6✔
127	if err != nil {	6✔
128	h.logger.Warn("websocket accept failed", "error", err)	×
129		×
130	return	×
131	}	×
132
133	canSend := h.canSendCommands(ctx, sess.User, server)	6✔
134		6✔
135	sessionID := idgen.New()	6✔
136	h.runAttachSession(ctx, conn, server, node, sessionID, sess.User, canSend)	6✔
137	}
138
139	func (h *Handler) runAttachSession(
140	ctx context.Context,
141	conn *websocket.Conn,
142	server *domain.Server,
143	node *domain.Node,
144	sessionID string,
145	user *domain.User,
146	canSend bool,
147	) {	6✔
148	client := ws.NewClient(ctx, conn, h.hub, nil, h.logger)	6✔
149		6✔
150	startedTopic := ws.ChannelToTopic(channels.BuildRealtimeAttachStartedChannel(sessionID))	6✔
151	outputTopic := ws.ChannelToTopic(channels.BuildRealtimeAttachOutputChannel(sessionID))	6✔
152	closedTopic := ws.ChannelToTopic(channels.BuildRealtimeAttachClosedChannel(sessionID))	6✔
153		6✔
154	h.hub.Register(client, startedTopic, outputTopic, closedTopic)	6✔
155	h.attachHandler.TrackAttachSession(sessionID, uint64(server.ID))	6✔
156		6✔
157	nodeID := uint64(node.ID)	6✔
158		6✔
159	msgHandler := h.newMessageHandler(ctx, client, server, node, user, sessionID, canSend)	6✔
160	client.SetMessageHandler(msgHandler)	6✔
161		6✔
162	if err := h.registry.SendAttachRequest(ctx, nodeID, &proto.AttachRequest{	6✔
163	SessionId: sessionID,	6✔
164	ServerId: uint64(server.ID),	6✔
165	}); err != nil {	6✔
166	h.logger.Warn("failed to send attach request",	×
167	"server_id", server.ID,	×
168	"session_id", sessionID,	×
169	"error", err,	×
170	)	×
171	client.SendMessage(ws.NewErrorMessage("failed to attach to server console"))	×
172	}	×
173
174	client.Run()	6✔
175		6✔
176	_ = h.registry.SendAttachDetach(context.Background(), nodeID, &proto.AttachDetach{	6✔
177	SessionId: sessionID,	6✔
178	Reason: "client disconnected",	6✔
179	})	6✔
180	h.attachHandler.UntrackAttachSession(sessionID)	6✔
181	}
182
183	func (h *Handler) newMessageHandler(
184	ctx context.Context,
185	client *ws.Client,
186	server *domain.Server,
187	node *domain.Node,
188	user *domain.User,
189	sessionID string,
190	canSend bool,
191	) ws.MessageHandler {	6✔
192	nodeID := uint64(node.ID)	6✔
193		6✔
194	return func(_ context.Context, msg *ws.InboundMessage) {	9✔
195	switch msg.Type {	3✔
196	case typeAttachInput:	2✔
197	if !canSend {	3✔
198	client.SendMessage(ws.NewErrorMessage("permission denied: cannot send input"))	1✔
199		1✔
200	return	1✔
201	}	1✔
202
203	if err := h.abilityChecker.CheckOrError(	1✔
204	ctx,	1✔
205	user.ID,	1✔
206	server.ID,	1✔
207	[]domain.AbilityName{domain.AbilityNameGameServerConsoleSend},	1✔
208	); err != nil {	1✔
209	client.SendMessage(ws.NewErrorMessage("permission denied: cannot send input"))	×
210		×
211	return	×
212	}	×
213
214	var payload attachInputPayload	1✔
215	if err := json.Unmarshal(msg.Payload, &payload); err != nil {	1✔
216	return	×
217	}	×
218
219	if err := h.registry.SendAttachInput(ctx, nodeID, &proto.AttachInput{	1✔
220	SessionId: sessionID,	1✔
221	Data: payload.Data,	1✔
222	}); err != nil {	1✔
223	h.logger.Warn("failed to send attach input",	×
224	"session_id", sessionID,	×
225	"error", err,	×
226	)	×
227	}	×
228
229	case typeAttachDetach:	1✔
230	_ = h.registry.SendAttachDetach(ctx, nodeID, &proto.AttachDetach{	1✔
231	SessionId: sessionID,	1✔
232	Reason: "user detached",	1✔
233	})	1✔
234	client.Close()	1✔
235	}
236	}
237	}
238
239	func (h Handler) findNode(ctx context.Context, nodeID uint) (domain.Node, error) {	7✔
240	nodes, err := h.nodeRepo.Find(ctx, &filters.FindNode{	7✔
241	IDs: []uint{nodeID},	7✔
242	}, nil, &filters.Pagination{	7✔
243	Limit: 1,	7✔
244	})	7✔
245	if err != nil {	7✔
246	return nil, errors.WithMessage(err, "failed to find node")	×
247	}	×
248
249	if len(nodes) == 0 {	8✔
250	return nil, api.NewNotFoundError("node not found")	1✔
251	}	1✔
252
253	return &nodes[0], nil	6✔
254	}
255
256	func (h Handler) canSendCommands(ctx context.Context, user domain.User, server *domain.Server) bool {	6✔
257	err := h.abilityChecker.CheckOrError(	6✔
258	ctx,	6✔
259	user.ID,	6✔
260	server.ID,	6✔
261	[]domain.AbilityName{domain.AbilityNameGameServerConsoleSend},	6✔
262	)	6✔
263		6✔
264	return err == nil	6✔
265	}	6✔
266
267	type attachInputPayload struct {
268	Data []byte `json:"data"`
269	}

gameap / gameap / 26129195806

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous