summaryrefslogtreecommitdiff
path: root/wsc.go
blob: 97e649a2bbdfdaa1c6c1e7c1e21515532de55fcc (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
/*
 * WebSocket connection routine
 *
 * Copyright (C) 2024  Runxi Yu <https://runxiyu.org>
 * SPDX-License-Identifier: AGPL-3.0-or-later
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <https://www.gnu.org/licenses/>.
 */

package main

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
	"time"

	"github.com/coder/websocket"
)

type errbytesT struct {
	err   error
	bytes *[]byte
}

var usemCount int64 /* atomic */

/*
 * This is more appropriately typed as uint64, but it needs to be cast to int64
 * later anyway due to time.Duration, so let's just use int64.
 */

/*
 * The actual logic in handling the connection, after authentication has been
 * completed.
 */
func handleConn(
	ctx context.Context,
	c *websocket.Conn,
	session string,
	userID string,
) (retErr error) {
	reportError := makeReportError(ctx, c)
	newCtx, newCancel := context.WithCancel(ctx)

	_cancel, ok := cancelPool.Load(userID)
	if ok {
		cancel, ok := _cancel.(*context.CancelFunc)
		if ok && cancel != nil {
			(*cancel)()
		}
		/* TODO: Make the cancel synchronous */
	}
	cancelPool.Store(userID, &newCancel)

	defer func() {
		cancelPool.CompareAndDelete(userID, &newCancel)
		if errors.Is(retErr, context.Canceled) {
			/*
			 * Only works if it's newCtx that has been cancelled
			 * rather than the original ctx, which is kinda what
			 * we intend
			 */
			_ = writeText(ctx, c, "E :Context canceled")
		}
		/* TODO: Report errors properly */
	}()

	/* TODO: Tell the user their current choices here. Deprecate HELLO. */

	usems := make(map[int]*usemT)

	/* TODO: Check if the LoadUint32 here is a bit too much overhead */
	atomic.AddInt64(&usemCount, int64(atomic.LoadUint32(&numCourses)))
	courses.Range(func(key, value interface{}) bool {
		/* TODO: Remember to change this too when changing the courseID type */
		courseID, ok := key.(int)
		if !ok {
			panic("courses map has non-\"int\" keys")
		}
		course, ok := value.(*courseT)
		if !ok {
			panic("courses map has non-\"*courseT\" items")
		}
		usem := &usemT{} //exhaustruct:ignore
		usem.init()
		course.Usems.Store(userID, usem)
		usems[courseID] = usem
		return true
	})

	defer func() {
		courses.Range(func(key, value interface{}) bool {
			_ = key
			course, ok := value.(*courseT)
			if !ok {
				panic("courses map has non-\"*courseT\" items")
			}
			course.Usems.Delete(userID)
			return true
		})
		atomic.AddInt64(&usemCount, -int64(atomic.LoadUint32(&numCourses)))
	}()

	usemParent := make(chan int)
	for courseID, usem := range usems {
		go func() {
			for {
				select {
				case <-newCtx.Done():
					return
				case <-usem.ch:
					select {
					case <-newCtx.Done():
						return
					case usemParent <- courseID:
					}
				}
				time.Sleep(
					time.Duration(
						atomic.LoadInt64(&usemCount)>>
							config.Perf.UsemDelayShiftBits,
					) * time.Millisecond,
				)
			}
		}()
	}

	/*
	 * userCourseGroups stores whether the user has already chosen a course
	 * in the courseGroup.
	 */
	var userCourseGroups userCourseGroupsT = make(map[courseGroupT]struct{})
	err := populateUserCourseGroups(newCtx, &userCourseGroups, userID)
	if err != nil {
		return reportError(
			fmt.Sprintf(
				"cannot populate user course groups: %v",
				err,
			),
		)
	}

	/*
	 * Later we need to select from recv and send and perform the
	 * corresponding action. But we can't just select from c.Read because
	 * the function blocks. Therefore, we must spawn a goroutine that
	 * blocks on c.Read and send what it receives to a channel "recv"; and
	 * then we can select from that channel.
	 */
	recv := make(chan *errbytesT)
	go func() {
		for {
			/*
			 * Here we use the original connection context instead
			 * of the new context we just created. Apparently when
			 * the context passed to Read expires, the connection
			 * gets closed, which makes it impossible for us to
			 * write the context expiry message to the client.
			 * So we pass the original connection context, which
			 * would get cancelled anyway once we close the
			 * connection.
			 * See: https://github.com/coder/websocket/issues/242
			 * We still need to take care of this while sending so
			 * we don't infinitely block, and leak goroutines and
			 * cause the channel to remain out of reach of the
			 * garbage collector.
			 * It would be nice to return the newCtx.Err() but
			 * the only way to really do that is to use the recv
			 * channel which might not have a listener anymore.
			 * It's not really crucial anyways so we could just
			 * close this goroutine by returning here.
			 */
			_, b, err := c.Read(ctx)
			if err != nil {
				/*
				 * TODO: Prioritize context dones... except
				 * that it's not really possible. I would just
				 * have placed newCtx in here but apparently
				 * that causes the connection to be closed when
				 * the context expires, which makes it
				 * impossible to deliver the final error
				 * message. Probably need to look into this
				 * design again.
				 */
				select {
				case <-newCtx.Done():
					_ = writeText(
						ctx,
						c,
						"E :Context canceled",
					)
					/* Not a typo to use ctx here */
					return
				case recv <- &errbytesT{err: err, bytes: nil}:
				}
				return
			}
			select {
			case <-newCtx.Done():
				_ = writeText(ctx, c, "E :Context cancelled")
				/* Not a typo to use ctx here */
				return
			case recv <- &errbytesT{err: nil, bytes: &b}:
			}
		}
	}()

	for {
		var mar []string
		select {
		case <-newCtx.Done():
			/*
			 * TODO: Somehow prioritize this case over all other
			 * cases
			 */
			return fmt.Errorf(
				"%w: %w",
				errContextCancelled,
				newCtx.Err(),
			)
			/*
			 * There are other times when the context could be
			 * cancelled, and apparently some WebSocket functions
			 * just close the connection when the context is
			 * cancelled. So it's kinda impossible to reliably
			 * send this message due to newCtx cancellation.
			 * But in any case, the WebSocket connection would
			 * be closed, and the user would see the connection
			 * closed page which should explain it.
			 */
		case courseID := <-usemParent:
			err := sendSelectedUpdate(newCtx, c, courseID)
			if err != nil {
				return fmt.Errorf(
					"%w: %w",
					errCannotSend,
					err,
				)
			}
			continue
		case errbytes := <-recv:
			if errbytes.err != nil {
				return fmt.Errorf(
					"%w: %w",
					errCannotReceiveMessage,
					errbytes.err,
				)
				/*
				 * Note that this cannot return newCtx.Err(),
				 * so we handle the error reporting in the
				 * reading routine
				 */
			}
			mar = splitMsg(errbytes.bytes)
			switch mar[0] {
			case "HELLO":
				err := messageHello(
					newCtx,
					c,
					reportError,
					mar,
					userID,
					session,
				)
				if err != nil {
					return err
				}
			case "Y":
				err := messageChooseCourse(
					newCtx,
					c,
					reportError,
					mar,
					userID,
					session,
					&userCourseGroups,
				)
				if err != nil {
					return err
				}
			case "N":
				err := messageUnchooseCourse(
					newCtx,
					c,
					reportError,
					mar,
					userID,
					session,
					&userCourseGroups,
				)
				if err != nil {
					return err
				}
			default:
				return reportError("Unknown command " + mar[0])
			}
		}
	}
}

var cancelPool sync.Map /* string, *context.CancelFunc */