First pass at adding a waiting room.

rtpconn/unlock.go

@@ -42,6 +42,20 @@ var hallUnlockSchedules = struct {
 	entries: make(map[*hall.Hall]*scheduledHallUnlock),
 }
 
+func scheduledUnlockDeadline(g *hall.Hall) (time.Time, bool) {
+	hallUnlockSchedules.Lock()
+	entry := hallUnlockSchedules.entries[g]
+	if entry == nil {
+		hallUnlockSchedules.Unlock()
+		return time.Time{}, false
+	}
+	entry.mu.Lock()
+	deadline, active := entry.deadline, entry.active
+	entry.mu.Unlock()
+	hallUnlockSchedules.Unlock()
+	return deadline, active
+}
+
 func scheduledUnlockRequestFromValue(value interface{}) (scheduledUnlockRequest, error) {
 	v, ok := value.(map[string]interface{})
 	if !ok {
@@ -265,6 +279,7 @@ func scheduleHallUnlock(g *hall.Hall, issuer *webClient, deadline, now time.Time
 		broadcastHallInfo(g, message, true)
 	}
 	entry.mu.Unlock()
+	waitingHallChanged(g)
 	return nil
 }
 
@@ -310,6 +325,7 @@ func finishScheduledUnlock(entry *scheduledHallUnlock) {
 	}
 	entry.mu.Unlock()
 	hallUnlockSchedules.Unlock()
+	waitingHallChanged(entry.hall)
 }
 
 func setHallLockState(g *hall.Hall, locked bool, message string) bool {
@@ -329,6 +345,7 @@ func setHallLockState(g *hall.Hall, locked bool, message string) bool {
 		broadcastHallInfo(g, "Hall unlocked", false)
 	}
 	hallUnlockSchedules.Unlock()
+	waitingHallChanged(g)
 	return entry != nil
 }
 
@@ -343,6 +360,21 @@ func cancelScheduledHallUnlock(g *hall.Hall) bool {
 	return entry != nil
 }
 
+func cancelScheduledUnlockByIssuer(g *hall.Hall, issuer *webClient) bool {
+	hallUnlockSchedules.Lock()
+	entry := hallUnlockSchedules.entries[g]
+	if entry == nil || entry.issuer != issuer {
+		hallUnlockSchedules.Unlock()
+		return false
+	}
+	delete(hallUnlockSchedules.entries, g)
+	deactivateScheduledUnlock(entry)
+	hallUnlockSchedules.Unlock()
+	broadcastHallInfo(g, "Scheduled hall unlock cancelled because the issuing operator is no longer present as an operator.", true)
+	waitingHallChanged(g)
+	return true
+}
+
 func deactivateScheduledUnlock(entry *scheduledHallUnlock) {
 	entry.mu.Lock()
 	entry.active = false

rtpconn/waiting.go

@@ -0,0 +1,330 @@
+package rtpconn
+
+import (
+	"errors"
+	"sync"
+	"time"
+
+	"git.stormux.org/storm/skald/hall"
+)
+
+type waitingEntry struct {
+	client     *webClient
+	hall       *hall.Hall
+	auth       hall.AuthenticatedClient
+	autoJoin   bool
+	approved   bool
+	manualJoin bool
+}
+
+type waitingEventAction struct {
+	message clientMessage
+}
+
+type waitingState struct {
+	Hall        string `json:"hall"`
+	DisplayName string `json:"displayName"`
+	Username    string `json:"username"`
+	Locked      bool   `json:"locked"`
+	Deadline    string `json:"deadline,omitempty"`
+}
+
+type waitingUser struct {
+	Username string `json:"username"`
+	Approved bool   `json:"approved,omitempty"`
+}
+
+var waitingRooms = struct {
+	sync.Mutex
+	entries map[*hall.Hall][]*waitingEntry
+}{
+	entries: make(map[*hall.Hall][]*waitingEntry),
+}
+
+var errWaitingUnavailable = errors.New("waiting room unavailable")
+
+func waitingDisplayName(g *hall.Hall) string {
+	name := g.Status(true, nil).DisplayName
+	if name == "" {
+		name = g.Name()
+	}
+	return name
+}
+
+func currentWaitingState(g *hall.Hall, username string) waitingState {
+	locked, _ := g.Locked()
+	state := waitingState{
+		Hall: g.Name(), DisplayName: waitingDisplayName(g),
+		Username: username, Locked: locked,
+	}
+	if deadline, ok := scheduledUnlockDeadline(g); ok {
+		state.Deadline = deadline.Format(time.RFC3339)
+	}
+	return state
+}
+
+func sendWaiting(c *webClient, kind string, value interface{}) {
+	c.action(waitingEventAction{clientMessage{
+		Type: "waiting", Kind: kind, Value: value,
+	}})
+}
+
+func sendWaitingList(c *webClient, kind string, value interface{}) {
+	c.action(waitingEventAction{clientMessage{
+		Type: "waiting-list", Kind: kind, Value: value,
+	}})
+}
+
+func waitingSnapshotLocked(g *hall.Hall) []waitingUser {
+	entries := waitingRooms.entries[g]
+	users := make([]waitingUser, 0, len(entries))
+	for _, entry := range entries {
+		users = append(users, waitingUser{entry.auth.Username, entry.approved})
+	}
+	return users
+}
+
+func waitingClientHall(c *webClient) *hall.Hall {
+	waitingRooms.Lock()
+	defer waitingRooms.Unlock()
+	if c.waiting == nil {
+		return nil
+	}
+	return c.waiting.hall
+}
+
+func notifyWaitingOperators(g *hall.Hall, kind string, value interface{}) {
+	for _, client := range g.GetClients(nil) {
+		c, ok := client.(*webClient)
+		if ok && member("op", c.Permissions()) {
+			sendWaitingList(c, kind, value)
+		}
+	}
+}
+
+func addWaitingClient(c *webClient, g *hall.Hall, auth hall.AuthenticatedClient) error {
+	waitingRooms.Lock()
+	offered, err := g.ReserveUsernameWhileWaiting(auth.Username, c)
+	if err != nil {
+		waitingRooms.Unlock()
+		return err
+	}
+	if !offered {
+		waitingRooms.Unlock()
+		return errWaitingUnavailable
+	}
+	entry := &waitingEntry{client: c, hall: g, auth: auth}
+	waitingRooms.entries[g] = append(waitingRooms.entries[g], entry)
+	c.waiting = entry
+	c.SetUsername(auth.Username)
+	sendWaiting(c, "enter", currentWaitingState(g, auth.Username))
+	notifyWaitingOperators(g, "add", waitingUser{Username: auth.Username})
+	waitingRooms.Unlock()
+	return nil
+}
+
+func removeWaitingClient(c *webClient, notify bool) bool {
+	waitingRooms.Lock()
+	entry := c.waiting
+	if entry == nil {
+		waitingRooms.Unlock()
+		return false
+	}
+	removeWaitingEntryLocked(entry)
+	if notify {
+		notifyWaitingOperators(entry.hall, "remove", waitingUser{Username: entry.auth.Username})
+	}
+	waitingRooms.Unlock()
+	return true
+}
+
+func removeWaitingEntryLocked(entry *waitingEntry) {
+	entries := waitingRooms.entries[entry.hall]
+	for i, candidate := range entries {
+		if candidate == entry {
+			entries = append(entries[:i], entries[i+1:]...)
+			break
+		}
+	}
+	if len(entries) == 0 {
+		delete(waitingRooms.entries, entry.hall)
+	} else {
+		waitingRooms.entries[entry.hall] = entries
+	}
+	entry.client.waiting = nil
+	entry.hall.ReleaseUsername(entry.auth.Username, entry.client)
+}
+
+func cancelWaiting(g *hall.Hall, message string) {
+	waitingRooms.Lock()
+	cancelWaitingLocked(g, message)
+	waitingRooms.Unlock()
+}
+
+func cancelWaitingLocked(g *hall.Hall, message string) {
+	entries := append([]*waitingEntry(nil), waitingRooms.entries[g]...)
+	delete(waitingRooms.entries, g)
+	for _, entry := range entries {
+		entry.client.waiting = nil
+		entry.hall.ReleaseUsername(entry.auth.Username, entry.client)
+		entry.client.SetPermissions(nil)
+		entry.client.SetUsername("")
+		sendWaiting(entry.client, "cancel", message)
+	}
+}
+
+func waitingOperatorJoined(g *hall.Hall, c *webClient) {
+	if !member("op", c.Permissions()) {
+		return
+	}
+	waitingRooms.Lock()
+	snapshot := waitingSnapshotLocked(g)
+	sendWaitingList(c, "snapshot", snapshot)
+	waitingRooms.Unlock()
+}
+
+func waitingHallChanged(g *hall.Hall) {
+	waitingRooms.Lock()
+	locked, hasOperator := g.WaitingStatus()
+	if locked && !hasOperator {
+		cancelWaitingLocked(g, "Waiting ended because the last operator left the hall.")
+		waitingRooms.Unlock()
+		return
+	}
+	entries := waitingRooms.entries[g]
+	if locked {
+		for _, entry := range entries {
+			entry.manualJoin = false
+		}
+	}
+	kind := "update"
+	if !locked {
+		kind = "open"
+	}
+	for _, entry := range entries {
+		sendWaiting(entry.client, kind, currentWaitingState(g, entry.auth.Username))
+	}
+	waitingRooms.Unlock()
+	processWaiting(g)
+}
+
+func processWaiting(g *hall.Hall) {
+	for {
+		waitingRooms.Lock()
+		entries := waitingRooms.entries[g]
+		locked, hasOperator := g.WaitingStatus()
+		if locked && !hasOperator {
+			cancelWaitingLocked(g, "Waiting ended because the last operator left the hall.")
+			waitingRooms.Unlock()
+			return
+		}
+		var entry *waitingEntry
+		for _, candidate := range entries {
+			if candidate.approved || (!locked && (candidate.autoJoin || candidate.manualJoin)) {
+				entry = candidate
+				break
+			}
+		}
+		if entry == nil {
+			waitingRooms.Unlock()
+			return
+		}
+		// Publish the target hall before AddAuthenticatedClient enqueues its
+		// joined callback; admission may run on a different client's goroutine.
+		entry.client.hall = g
+		err := hall.AddAuthenticatedClient(g, entry.client, entry.auth, entry.approved)
+		if err != nil {
+			entry.client.hall = nil
+			if lockedNow, _ := g.Locked(); lockedNow && !entry.approved {
+				entry.manualJoin = false
+				waitingRooms.Unlock()
+				sendWaiting(entry.client, "update", currentWaitingState(g, entry.auth.Username))
+				return
+			}
+			if errors.Is(err, hall.UserError("too many users")) || err.Error() == "too many users" {
+				waitingRooms.Unlock()
+				return
+			}
+			removeWaitingEntryLocked(entry)
+			entry.client.SetPermissions(nil)
+			entry.client.SetUsername("")
+			notifyWaitingOperators(g, "remove", waitingUser{Username: entry.auth.Username})
+			waitingRooms.Unlock()
+			sendWaiting(entry.client, "cancel", "Admission cannot continue: "+err.Error())
+			continue
+		}
+		removeWaitingEntryLocked(entry)
+		notifyWaitingOperators(g, "remove", waitingUser{Username: entry.auth.Username})
+		waitingRooms.Unlock()
+		waitingOperatorJoined(g, entry.client)
+	}
+}
+
+// WaitingCapacityChanged retries pending admissions after a non-WebSocket
+// client leaves the hall.
+func WaitingCapacityChanged(g *hall.Hall) {
+	processWaiting(g)
+}
+
+func updateWaitingPreference(c *webClient, autoJoin bool) error {
+	waitingRooms.Lock()
+	if c.waiting == nil {
+		waitingRooms.Unlock()
+		return hall.UserError("you are not in a waiting room")
+	}
+	c.waiting.autoJoin = autoJoin
+	g := c.waiting.hall
+	waitingRooms.Unlock()
+	if autoJoin {
+		processWaiting(g)
+	}
+	return nil
+}
+
+func requestWaitingAdmission(c *webClient) error {
+	waitingRooms.Lock()
+	if c.waiting == nil {
+		waitingRooms.Unlock()
+		return hall.UserError("you are not in a waiting room")
+	}
+	g := c.waiting.hall
+	waitingRooms.Unlock()
+	locked, _ := g.Locked()
+	if locked {
+		return hall.UserError("the hall is still locked")
+	}
+	waitingRooms.Lock()
+	if c.waiting != nil {
+		c.waiting.manualJoin = true
+	}
+	waitingRooms.Unlock()
+	processWaiting(g)
+	return nil
+}
+
+func admitWaitingClient(operator *webClient, username string) error {
+	if operator.hall == nil || !member("op", operator.Permissions()) {
+		return hall.UserError("not authorised")
+	}
+	g := operator.hall
+	waitingRooms.Lock()
+	var found *waitingEntry
+	for _, entry := range waitingRooms.entries[g] {
+		if entry.auth.Username == username {
+			entry.approved = true
+			found = entry
+			break
+		}
+	}
+	snapshot := waitingSnapshotLocked(g)
+	if found != nil {
+		notifyWaitingOperators(g, "snapshot", snapshot)
+	}
+	waitingRooms.Unlock()
+	if found == nil {
+		return hall.UserError("no such waiting user")
+	}
+	sendWaiting(found.client, "approved", currentWaitingState(g, username))
+	processWaiting(g)
+	return nil
+}

rtpconn/waiting_test.go

@@ -0,0 +1,315 @@
+package rtpconn
+
+import (
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"git.stormux.org/storm/skald/hall"
+)
+
+func newWaitingTestHall(t *testing.T, maxClients int) (*hall.Hall, string) {
+	t.Helper()
+	name := "waiting-" + strings.NewReplacer("/", "-", " ", "-").Replace(t.Name())
+	oldDirectory := hall.Directory
+	hall.Directory = t.TempDir()
+	t.Cleanup(func() { hall.Directory = oldDirectory })
+	password := testPassword()
+	desc := &hall.Description{
+		DisplayName: "Stormux", MaxClients: maxClients,
+		Users: map[string]hall.UserDescription{
+			"Operator":  {Password: password, Permissions: makePermission(t, "op")},
+			"Operator2": {Password: password, Permissions: makePermission(t, "op")},
+			"Alice":     {Password: password, Permissions: makePermission(t, "present")},
+			"Bob":       {Password: password, Permissions: makePermission(t, "present")},
+		},
+	}
+	data, err := json.Marshal(desc)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(hall.Directory, name+".json"), data, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	g, err := hall.Add(name, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		cancelScheduledHallUnlock(g)
+		cancelWaiting(g, "test cleanup")
+		for _, client := range g.GetClients(nil) {
+			hall.DelClient(client)
+		}
+		hall.Delete(name)
+	})
+	return g, name
+}
+
+func joinWaitingTestClient(t *testing.T, name, username string) *webClient {
+	t.Helper()
+	c := testWebClient(username + "-waiting-id")
+	err := handleClientMessage(c, clientMessage{
+		Type: "join", Kind: "join", Hall: name,
+		Username: &username, Password: "pw",
+	})
+	if err != nil {
+		t.Fatalf("waiting join: %v", err)
+	}
+	drainActions(t, c)
+	return c
+}
+
+func TestWaitingClientAuthenticatedWithoutMembership(t *testing.T) {
+	g, name := newWaitingTestHall(t, 0)
+	operator := addTestWebClient(t, name, "Operator", "op")
+	g.SetLocked(true, "")
+	drainActions(t, operator)
+
+	waiter := joinWaitingTestClient(t, name, "Alice")
+	if waiter.hall != nil || waiter.waiting == nil {
+		t.Fatalf("waiting client membership: hall=%v waiting=%v", waiter.hall, waiter.waiting)
+	}
+	if waiter.Username() != "Alice" || waiter.waiting.auth.Username != "Alice" {
+		t.Fatalf("authenticated identity was not retained: %#v", waiter.waiting.auth)
+	}
+	if g.GetClient(waiter.Id()) != nil {
+		t.Fatal("waiting client was visible as a hall member")
+	}
+	if err := handleClientMessage(waiter, clientMessage{Type: "chat", Value: "private"}); err != nil {
+		t.Fatalf("isolated chat attempt returned fatal error: %v", err)
+	}
+	if len(g.GetChatHistory()) != 0 {
+		t.Fatal("waiting client wrote to hall chat")
+	}
+	duplicate := testWebClient("direct-duplicate-id")
+	username := "Alice"
+	if _, err := hall.AddClient(name, duplicate, hall.ClientCredentials{
+		Username: &username, Password: "pw",
+	}); err == nil {
+		t.Fatal("waiting username was not reserved against direct admission")
+	}
+	messages := drainMessages(waiter)
+	foundEnter := false
+	for _, message := range messages {
+		if message.Type == "waiting" && message.Kind == "enter" {
+			foundEnter = true
+		}
+		if message.Type == "user" || message.Type == "chat" || message.Type == "chathistory" || message.Type == "offer" ||
+			(message.Type == "usermessage" && (message.Kind == "recording" || message.Kind == "chalkboard")) {
+			t.Fatalf("waiting client received hall-only message: %#v", message)
+		}
+	}
+	if !foundEnter {
+		t.Fatalf("missing waiting enter event: %#v", messages)
+	}
+}
+
+func TestWaitingManualAdmissionAndLastOperatorCancellation(t *testing.T) {
+	g, name := newWaitingTestHall(t, 0)
+	operator := addTestWebClient(t, name, "Operator", "op")
+	g.SetLocked(true, "")
+	drainActions(t, operator)
+	waiter := joinWaitingTestClient(t, name, "Alice")
+
+	if err := admitWaitingClient(operator, "Alice"); err != nil {
+		t.Fatal(err)
+	}
+	drainActions(t, waiter)
+	if waiter.hall != g || waiter.waiting != nil {
+		t.Fatalf("operator admission did not join: hall=%v waiting=%v", waiter.hall, waiter.waiting)
+	}
+
+	second := joinWaitingTestClient(t, name, "Bob")
+	leaveHall(operator)
+	drainActions(t, second)
+	if second.waiting != nil {
+		t.Fatal("waiting entry remained after last operator left")
+	}
+	if second.Username() != "" || len(second.Permissions()) != 0 {
+		t.Fatal("cancelled waiting client retained its authenticated identity")
+	}
+	foundCancel := false
+	for _, message := range drainMessages(second) {
+		if message.Type == "waiting" && message.Kind == "cancel" {
+			foundCancel = true
+		}
+	}
+	if !foundCancel {
+		t.Fatal("waiting client did not receive cancellation")
+	}
+}
+
+func TestWaitingAutomaticAdmissionRetriesInOrderAtCapacity(t *testing.T) {
+	g, name := newWaitingTestHall(t, 2)
+	operator := addTestWebClient(t, name, "Operator", "op")
+	g.SetLocked(true, "")
+	drainActions(t, operator)
+	alice := joinWaitingTestClient(t, name, "Alice")
+	bob := joinWaitingTestClient(t, name, "Bob")
+	if err := updateWaitingPreference(alice, true); err != nil {
+		t.Fatal(err)
+	}
+	if err := updateWaitingPreference(bob, true); err != nil {
+		t.Fatal(err)
+	}
+	setHallLockState(g, false, "")
+	drainActions(t, alice)
+	drainActions(t, bob)
+	if alice.hall != g || bob.hall != nil || bob.waiting == nil {
+		t.Fatalf("unexpected first admission: alice=%v bob=%v bobWaiting=%v", alice.hall, bob.hall, bob.waiting)
+	}
+	leaveHall(alice)
+	drainActions(t, bob)
+	if bob.hall != g || bob.waiting != nil {
+		t.Fatalf("capacity retry did not admit Bob: hall=%v waiting=%v", bob.hall, bob.waiting)
+	}
+}
+
+func TestWaitingAdmissionRequiresOperator(t *testing.T) {
+	g, name := newWaitingTestHall(t, 0)
+	operator := addTestWebClient(t, name, "Operator", "op")
+	g.SetLocked(true, "")
+	drainActions(t, operator)
+	alice := joinWaitingTestClient(t, name, "Alice")
+	if err := admitWaitingClient(alice, "Alice"); err == nil {
+		t.Fatal("waiting user admitted themself")
+	}
+}
+
+func TestWaitingRequiresOperatorAndReservesUsername(t *testing.T) {
+	g, name := newWaitingTestHall(t, 0)
+	g.SetLocked(true, "")
+	alice := joinWaitingTestClient(t, name, "Alice")
+	if alice.waiting != nil {
+		t.Fatal("client waited without an active operator")
+	}
+	foundFailure := false
+	for _, message := range drainMessages(alice) {
+		if message.Type == "joined" && message.Kind == "fail" {
+			foundFailure = true
+		}
+	}
+	if !foundFailure {
+		t.Fatal("locked hall without an operator did not reject the join")
+	}
+
+	operator := addTestWebClient(t, name, "Operator", "op")
+	drainActions(t, operator)
+	first := joinWaitingTestClient(t, name, "Alice")
+	second := joinWaitingTestClient(t, name, "Alice")
+	if first.waiting == nil || second.waiting != nil {
+		t.Fatalf("username reservation failed: first=%v second=%v", first.waiting, second.waiting)
+	}
+	removeWaitingClient(first, true)
+	third := joinWaitingTestClient(t, name, "Alice")
+	if third.waiting == nil {
+		t.Fatal("username was not released after waiting client departed")
+	}
+}
+
+func TestWaitingRelockDisablesManualAdmission(t *testing.T) {
+	g, name := newWaitingTestHall(t, 1)
+	operator := addTestWebClient(t, name, "Operator", "op")
+	g.SetLocked(true, "")
+	drainActions(t, operator)
+	alice := joinWaitingTestClient(t, name, "Alice")
+	setHallLockState(g, false, "")
+	setHallLockState(g, true, "")
+	if err := requestWaitingAdmission(alice); err == nil {
+		t.Fatal("manual admission succeeded after the hall was relocked")
+	}
+	if alice.hall != nil || alice.waiting == nil {
+		t.Fatalf("relocked client left waiting state: hall=%v waiting=%v", alice.hall, alice.waiting)
+	}
+}
+
+func TestWaitingManualJoinAfterOpening(t *testing.T) {
+	g, name := newWaitingTestHall(t, 0)
+	operator := addTestWebClient(t, name, "Operator", "op")
+	g.SetLocked(true, "")
+	drainActions(t, operator)
+	alice := joinWaitingTestClient(t, name, "Alice")
+	setHallLockState(g, false, "")
+	if err := requestWaitingAdmission(alice); err != nil {
+		t.Fatal(err)
+	}
+	drainActions(t, alice)
+	if alice.hall != g || alice.waiting != nil {
+		t.Fatalf("manual join did not complete: hall=%v waiting=%v", alice.hall, alice.waiting)
+	}
+}
+
+func TestWaitingApprovalSurvivesCapacityAndRelock(t *testing.T) {
+	g, name := newWaitingTestHall(t, 2)
+	operator := addTestWebClient(t, name, "Operator", "op")
+	g.SetLocked(true, "")
+	drainActions(t, operator)
+	alice := joinWaitingTestClient(t, name, "Alice")
+	bob := joinWaitingTestClient(t, name, "Bob")
+	if err := admitWaitingClient(operator, "Alice"); err != nil {
+		t.Fatal(err)
+	}
+	drainActions(t, alice)
+	if alice.hall != g {
+		t.Fatal("first approved user did not fill available capacity")
+	}
+	if err := admitWaitingClient(operator, "Bob"); err != nil {
+		t.Fatal(err)
+	}
+	if bob.waiting == nil || !bob.waiting.approved {
+		t.Fatal("approval was not retained while the hall was full")
+	}
+	setHallLockState(g, false, "")
+	setHallLockState(g, true, "")
+	leaveHall(alice)
+	drainActions(t, bob)
+	if bob.hall != g || bob.waiting != nil {
+		t.Fatalf("approved user was not admitted after capacity opened: hall=%v waiting=%v", bob.hall, bob.waiting)
+	}
+}
+
+func TestWaitingCancelledWhenLastOperatorLosesPermission(t *testing.T) {
+	g, name := newWaitingTestHall(t, 0)
+	operator := addTestWebClient(t, name, "Operator", "op")
+	g.SetLocked(true, "")
+	drainActions(t, operator)
+	alice := joinWaitingTestClient(t, name, "Alice")
+	if err := handleAction(operator, changePermissionsAction{kind: "unop"}); err != nil {
+		t.Fatal(err)
+	}
+	// Exercise the window before permissionsChangedAction is processed.
+	WaitingCapacityChanged(g)
+	drainActions(t, operator)
+	drainActions(t, alice)
+	if alice.waiting != nil {
+		t.Fatal("waiting continued after the last operator lost permission")
+	}
+}
+
+func TestWaitingScheduleRemovedWhenIssuerLeaves(t *testing.T) {
+	g, name := newWaitingTestHall(t, 0)
+	issuer := addTestWebClient(t, name, "Operator", "op")
+	other := addTestWebClient(t, name, "Operator2", "op")
+	g.SetLocked(true, "")
+	drainActions(t, issuer)
+	drainActions(t, other)
+	waiter := joinWaitingTestClient(t, name, "Alice")
+	now := unlockNow()
+	if err := scheduleHallUnlock(g, issuer, now.Add(10*time.Minute), now); err != nil {
+		t.Fatal(err)
+	}
+	if _, ok := scheduledUnlockDeadline(g); !ok {
+		t.Fatal("scheduled deadline was not recorded")
+	}
+	leaveHall(issuer)
+	if _, ok := scheduledUnlockDeadline(g); ok {
+		t.Fatal("scheduled deadline remained after issuer departure")
+	}
+	if waiter.waiting == nil {
+		t.Fatal("waiter was cancelled even though another operator remained")
+	}
+}

rtpconn/webclient.go

@@ -127,6 +127,8 @@ type webClient struct {
 	writerDone    chan struct{}
 	actions       *unbounded.Channel[any]
 	permissionsMu sync.RWMutex
+	identityMu    sync.RWMutex
+	waiting       *waitingEntry
 
 	mu   sync.Mutex
 	down map[string]*rtpDownConnection
@@ -146,11 +148,15 @@ func (c *webClient) Id() string {
 }
 
 func (c *webClient) Username() string {
+	c.identityMu.RLock()
+	defer c.identityMu.RUnlock()
 	return c.username
 }
 
 func (c *webClient) SetUsername(username string) {
+	c.identityMu.Lock()
 	c.username = username
+	c.identityMu.Unlock()
 }
 
 func (c *webClient) Permissions() []string {
@@ -857,7 +863,7 @@ func readMessage(conn *websocket.Conn, m *clientMessage) error {
 }
 
 const maxWSMessageSize = 1024 * 1024
-const protocolVersion = "2"
+const protocolVersion = "3"
 
 func StartClient(conn *websocket.Conn, addr net.Addr) (err error) {
 	var m clientMessage
@@ -1187,6 +1193,9 @@ func handleAction(c *webClient, a any) error {
 		}
 
 	case pushClientAction:
+		// Admission may be running on another client's goroutine.  Taking
+		// the waiting lock is a barrier for c.hall and registry publication.
+		waitingClientHall(c)
 		if a.hall != c.hall.Name() {
 			log.Printf("got client for wrong hall")
 			return nil
@@ -1202,6 +1211,7 @@ func handleAction(c *webClient, a any) error {
 			Data:        a.data,
 		})
 	case joinedAction:
+		waitingClientHall(c)
 		var status *hall.Status
 		var data map[string]interface{}
 		var g *hall.Hall
@@ -1215,7 +1225,7 @@ func handleAction(c *webClient, a any) error {
 			}
 		}
 		perms := append([]string(nil), c.permissions...)
-		username := c.username
+		username := c.Username()
 		err := c.write(clientMessage{
 			Type:             "joined",
 			Kind:             a.kind,
@@ -1295,9 +1305,14 @@ func handleAction(c *webClient, a any) error {
 			return errors.New("Permissions changed in no hall")
 		}
 		perms := c.Permissions()
+		if !member("op", perms) {
+			cancelScheduledUnlockByIssuer(g, c)
+		} else {
+			waitingOperatorJoined(g, c)
+		}
 		status := g.Status(true, nil)
 		status.Recording = hallRecording(g)
-		username := c.username
+		username := c.Username()
 		c.write(clientMessage{
 			Type:             "joined",
 			Kind:             "change",
@@ -1332,12 +1347,15 @@ func handleAction(c *webClient, a any) error {
 				)
 			}
 		}(clients)
+		waitingHallChanged(g)
 	case kickAction:
 		return hall.KickError{
 			Id:       a.id,
 			Username: a.username,
 			Message:  a.message,
 		}
+	case waitingEventAction:
+		return c.write(a.message)
 	default:
 		log.Printf("unexpected action %T", a)
 		return errors.New("unexpected action")
@@ -1365,6 +1383,11 @@ func failUpConnection(c *webClient, id string, message string) error {
 }
 
 func leaveHall(c *webClient) {
+	if removeWaitingClient(c, true) {
+		c.SetPermissions(nil)
+		c.SetUsername("")
+		return
+	}
 	if c.hall == nil {
 		return
 	}
@@ -1380,11 +1403,14 @@ func leaveHall(c *webClient) {
 		}
 	}
 
+	g := c.hall
+	cancelScheduledUnlockByIssuer(g, c)
 	hall.DelClient(c)
 	c.SetPermissions(nil)
 	c.data = nil
 	c.requested = make(map[string][]string)
 	c.hall = nil
+	waitingHallChanged(g)
 }
 
 func closeDownConn(c *webClient, id string, message string) error {
@@ -1441,10 +1467,19 @@ func handleClientMessage(c *webClient, m clientMessage) error {
 			}
 		}
 	}
+	if waitingClientHall(c) != nil && m.Type != "waiting" && m.Type != "join" && m.Type != "pong" {
+		return c.error(hall.UserError("you have not been admitted to the hall"))
+	}
 
 	switch m.Type {
 	case "join":
 		if m.Kind == "leave" {
+			if waitingHall := waitingClientHall(c); waitingHall != nil && waitingHall.Name() == m.Hall {
+				removeWaitingClient(c, true)
+				c.SetUsername("")
+				c.SetPermissions(nil)
+				return c.write(clientMessage{Type: "waiting", Kind: "leave", Hall: m.Hall})
+			}
 			if c.hall == nil || c.hall.Name() != m.Hall {
 				return hall.UserError("you are not joined")
 			}
@@ -1456,13 +1491,13 @@ func handleClientMessage(c *webClient, m clientMessage) error {
 			return hall.ProtocolError("unknown kind")
 		}
 
-		if c.hall != nil {
+		if c.hall != nil || waitingClientHall(c) != nil {
 			return hall.ProtocolError(
 				"cannot join multiple halls",
 			)
 		}
 		c.data = m.Data
-		g, err := hall.AddClient(m.Hall, c,
+		g, auth, err := hall.AuthenticateClient(m.Hall,
 			hall.ClientCredentials{
 				Username: m.Username,
 				Password: m.Password,
@@ -1490,7 +1525,7 @@ func handleClientMessage(c *webClient, m clientMessage) error {
 				s = "internal server error"
 				log.Printf("Join hall: %v", err)
 			}
-			username := c.username
+			username := c.Username()
 			return c.write(clientMessage{
 				Type:     "joined",
 				Kind:     "fail",
@@ -1503,7 +1538,7 @@ func handleClientMessage(c *webClient, m clientMessage) error {
 		if redirect := g.Description().Redirect; redirect != "" {
 			// We normally redirect at the HTTP level, but the hall
 			// description could have been edited in the meantime.
-			username := c.username
+			username := c.Username()
 			return c.write(clientMessage{
 				Type:     "joined",
 				Kind:     "redirect",
@@ -1512,7 +1547,60 @@ func handleClientMessage(c *webClient, m clientMessage) error {
 				Value:    redirect,
 			})
 		}
+		if !member("op", auth.Permissions) {
+			if err := addWaitingClient(c, g, auth); err == nil {
+				return nil
+			} else if !errors.Is(err, errWaitingUnavailable) {
+				return c.write(clientMessage{
+					Type: "joined", Kind: "fail", Hall: m.Hall,
+					Username: &auth.Username, Value: err.Error(),
+				})
+			}
+		}
+		if err := hall.AddAuthenticatedClient(g, c, auth, false); err != nil {
+			username := auth.Username
+			return c.write(clientMessage{
+				Type: "joined", Kind: "fail", Hall: m.Hall,
+				Username: &username, Value: err.Error(),
+			})
+		}
 		c.hall = g
+		waitingOperatorJoined(g, c)
+	case "waiting":
+		switch m.Kind {
+		case "auto":
+			autoJoin, ok := m.Value.(bool)
+			if !ok {
+				return c.error(hall.UserError("invalid automatic join preference"))
+			}
+			if err := updateWaitingPreference(c, autoJoin); err != nil {
+				return c.error(err)
+			}
+			return nil
+		case "join":
+			if err := requestWaitingAdmission(c); err != nil {
+				return c.error(err)
+			}
+			return nil
+		case "leave":
+			if !removeWaitingClient(c, true) {
+				return c.error(hall.UserError("you are not in a waiting room"))
+			}
+			c.SetUsername("")
+			c.SetPermissions(nil)
+			return c.write(clientMessage{Type: "waiting", Kind: "leave"})
+		case "admit":
+			username, ok := m.Value.(string)
+			if !ok || username == "" {
+				return c.error(hall.UserError("invalid waiting username"))
+			}
+			if err := admitWaitingClient(c, username); err != nil {
+				return c.error(err)
+			}
+			return nil
+		default:
+			return hall.ProtocolError("unknown waiting action")
+		}
 	case "request":
 		requested, err := parseRequested(m.Request)
 		if err != nil {
@@ -1877,7 +1965,7 @@ func handleClientMessage(c *webClient, m clientMessage) error {
 				}
 			}
 
-			user := c.username
+			user := c.Username()
 			if user != "" {
 				tok.IssuedBy = &user
 			}

rtpconn/whipclient.go

@@ -137,6 +137,7 @@ func (c *WhipClient) Close() error {
 		c.connection = nil
 	}
 	hall.DelClient(c)
+	WaitingCapacityChanged(g)
 	c.hall = nil
 	return nil
 }