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 }