skald/rtpconn/waiting.go

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
}