package meetroom /* MeetRoom - Video conferencing room manager author: tobychui / AI assisted This package implements the server-side room state for the MeetRoom video conferencing WebApp: meeting rooms joinable by ID + optional password, the participant registry used by the WebSocket signaling relay, an attendance log of every join / leave, and temporary attachment storage for in-meeting file sharing. When a sharedspace.Manager is bound (BindSpaceManager), every room created afterwards owns a shared space: chat messages and uploaded attachments are mirrored into it (origin OriginMeetRoom) so AGI scripts can read the meeting content, and items posted into the space from outside the room (e.g. by AGI scripts) are handed to the transport layer through SetSpaceItemHandler so they appear in the meeting live. The space is deleted together with the room. The package is transport-agnostic: the HTTP / WebSocket handlers live in the main package (src/meetroom.go) and only push byte slices into each participant's send channel. Media itself never touches the server - clients exchange WebRTC offers through the signaling relay and stream peer-to-peer. */ import ( "crypto/rand" "crypto/sha256" "crypto/subtle" "encoding/hex" "errors" "fmt" "io" mathrand "math/rand" "os" "path/filepath" "sync" "time" "imuslab.com/arozos/mod/sharedspace" ) var ( //ErrRoomNotFound is returned when the requested room ID does not exist ErrRoomNotFound = errors.New("room not found") //ErrInvalidPassword is returned when the room password does not match ErrInvalidPassword = errors.New("invalid room password") //ErrRoomClosed is returned when operating on a room that has been closed ErrRoomClosed = errors.New("room closed") //ErrAttachmentTooLarge is returned when an uploaded attachment exceeds the size limit ErrAttachmentTooLarge = errors.New("attachment exceeds size limit") ) const ( roomIDLength = 9 // digits in a meeting room ID, Zoom style maxTitleLength = 64 // room title is clipped to this many runes maxNameLength = 128 // attachment file names are clipped to this many runes maxAttendance = 1000 // attendance records kept per room (oldest dropped) DefaultMaxUpload = 128 << 20 // 128MB per attachment //OriginMeetRoom tags shared-space items mirrored from the room itself so //the space item bridge can filter its own echoes OriginMeetRoom = "meetroom" DefaultEmptyIdle = 10 * time.Minute ) // Participant is one connected member of a room. Since the migration onto // the sharedspace channel layer it is a thin wrapper around a channel // subscriber: PeerID and Send alias the subscriber's ID and send buffer. // The transport layer (WebSocket handler) drains Send and writes each frame // to the socket. type Participant struct { PeerID int Username string IsHost bool Send chan []byte // same channel value as sub.Send joinedAt time.Time sub *sharedspace.Subscriber } // CloseSend closes the participant's send channel exactly once (delegated to // the underlying subscriber, which owns the close-once guarantee). func (p *Participant) CloseSend() { p.sub.CloseSend() } // Attachment is a file shared into a room, stored on local disk until the // room is closed. Files are addressed by a random ID so the original file // name never becomes part of a filesystem path. type Attachment struct { ID string Name string Size int64 Uploader string DiskPath string } // AttendanceRecord is one join / leave entry in a room's attendance log. // LeftAt is the zero time while the participant is still in the meeting. type AttendanceRecord struct { Username string PeerID int JoinedAt time.Time LeftAt time.Time } // Present reports whether this record's participant is still in the room. func (a *AttendanceRecord) Present() bool { return a.LeftAt.IsZero() } // Room is one live meeting room. The realtime transport (participant // registry, broadcast, targeted send) rides the room's sharedspace channel: // for space-bound rooms that hub is shared with generic space subscribers, // so AGI-facing clients on the space receive meeting frames live. The // participants map is the meeting roster proper (host flag, attendance // identity) and is always a subset of the channel's subscribers. type Room struct { ID string Title string Host string SpaceID string // bound shared space, empty when no space manager is set CreatedAt time.Time passwordHash []byte // nil when the room has no password participants map[int]*Participant attachments map[string]*Attachment attendance []*AttendanceRecord channel *sharedspace.Channel // set once in CreateRoom, immutable afterwards lastActivity time.Time closed bool mu sync.Mutex } // Manager owns all live rooms and their attachment storage. type Manager struct { rooms map[string]*Room storageRoot string spaces *sharedspace.Manager // optional shared-space binding onSpaceItem func(room *Room, item *sharedspace.Item) // bridge for externally posted space items mu sync.RWMutex } // NewManager creates a room manager. storageRoot is the directory used for // temporary attachment storage; pass "" to use a folder inside os.TempDir(). // Any leftover attachment files from a previous run are removed. func NewManager(storageRoot string) *Manager { if storageRoot == "" { storageRoot = filepath.Join(os.TempDir(), "arozos", "meetroom") } //Attachments never survive a restart: rooms are in-memory only os.RemoveAll(storageRoot) os.MkdirAll(storageRoot, 0755) return &Manager{ rooms: make(map[string]*Room), storageRoot: storageRoot, } } // BindSpaceManager links the manager to a shared-space manager. Every room // created afterwards owns a shared space that mirrors its chat and // attachments and accepts posts from AGI scripts. func (m *Manager) BindSpaceManager(sm *sharedspace.Manager) { m.mu.Lock() defer m.mu.Unlock() m.spaces = sm } // SetSpaceItemHandler sets the callback invoked when an item lands in a // room's shared space from outside the room itself (anything whose origin is // not OriginMeetRoom, e.g. an AGI script). The transport layer uses it to // push the item into the meeting as a live chat / file message. func (m *Manager) SetSpaceItemHandler(fn func(room *Room, item *sharedspace.Item)) { m.mu.Lock() defer m.mu.Unlock() m.onSpaceItem = fn } // spaceOf returns the shared space bound to the room, if any. func (m *Manager) spaceOf(room *Room) (*sharedspace.Space, bool) { m.mu.RLock() sm := m.spaces m.mu.RUnlock() if sm == nil || room.SpaceID == "" { return nil, false } return sm.GetSpace(room.SpaceID) } // hashRoomPassword derives the stored hash for a room password. The room ID // acts as the salt so identical passwords in different rooms hash differently. func hashRoomPassword(roomID string, password string) []byte { sum := sha256.Sum256([]byte(roomID + ":" + password)) return sum[:] } // clipString trims s to at most max runes. func clipString(s string, max int) string { runes := []rune(s) if len(runes) > max { return string(runes[:max]) } return s } // CreateRoom creates a new room hosted by host. password may be empty for an // open room. The generated room ID is unique among live rooms. func (m *Manager) CreateRoom(host string, title string, password string) *Room { m.mu.Lock() defer m.mu.Unlock() var id string for { id = randomDigits(roomIDLength) if _, exists := m.rooms[id]; !exists { break } } title = clipString(title, maxTitleLength) if title == "" { title = host + "'s Meeting" } room := &Room{ ID: id, Title: title, Host: host, CreatedAt: time.Now(), participants: make(map[int]*Participant), attachments: make(map[string]*Attachment), attendance: []*AttendanceRecord{}, lastActivity: time.Now(), } if password != "" { room.passwordHash = hashRoomPassword(id, password) } //Bind a shared space to the room: chat / attachments mirror into it, //externally posted items (AGI) flow back through the item bridge, and //the space's realtime channel carries the meeting signaling. if m.spaces != nil { space := m.spaces.CreateSpace(host, title) room.SpaceID = space.ID room.channel = space.Channel() space.Subscribe(OriginMeetRoom, func(item *sharedspace.Item) { if item.Origin == OriginMeetRoom { return //the room's own echo, already delivered over WebSocket } m.mu.RLock() handler := m.onSpaceItem m.mu.RUnlock() if handler != nil { handler(room, item) } }) } else { //No space manager bound (e.g. unit tests): the room still needs a //realtime hub of its own room.channel = sharedspace.NewStandaloneChannel() } m.rooms[id] = room return room } // randomDigits returns n random decimal digits with no leading zero. func randomDigits(n int) string { digits := make([]byte, n) digits[0] = byte('1' + mathrand.Intn(9)) for i := 1; i < n; i++ { digits[i] = byte('0' + mathrand.Intn(10)) } return string(digits) } // GetRoom returns the live room with the given ID. func (m *Manager) GetRoom(id string) (*Room, bool) { m.mu.RLock() defer m.mu.RUnlock() room, ok := m.rooms[id] return room, ok } // ValidateJoin checks that the room exists and that the supplied password is // correct, returning the room on success. func (m *Manager) ValidateJoin(id string, password string) (*Room, error) { room, ok := m.GetRoom(id) if !ok { return nil, ErrRoomNotFound } if !room.CheckPassword(password) { return nil, ErrInvalidPassword } return room, nil } // HasPassword reports whether the room requires a password to join. func (r *Room) HasPassword() bool { r.mu.Lock() defer r.mu.Unlock() return r.passwordHash != nil } // CheckPassword reports whether the supplied password unlocks the room. func (r *Room) CheckPassword(password string) bool { r.mu.Lock() defer r.mu.Unlock() if r.passwordHash == nil { return true } candidate := hashRoomPassword(r.ID, password) return subtle.ConstantTimeCompare(r.passwordHash, candidate) == 1 } // AddParticipant joins the room's channel as a new subscriber, registers it // in the meeting roster and returns the wrapping participant. The transport // layer must drain the returned participant's Send channel. func (r *Room) AddParticipant(username string) (*Participant, error) { r.mu.Lock() if r.closed { r.mu.Unlock() return nil, ErrRoomClosed } r.mu.Unlock() //Rooms bind open spaces, so the channel ACL always admits; a closed //channel means the room was torn down while we were joining. sub, err := r.channel.Join(username) if err != nil { return nil, ErrRoomClosed } p := &Participant{ PeerID: sub.ID, Username: username, IsHost: username == r.Host, Send: sub.Send, joinedAt: sub.JoinedAt(), sub: sub, } r.mu.Lock() if r.closed { //The room closed between the pre-check and the channel join r.mu.Unlock() r.channel.Leave(sub.ID) return nil, ErrRoomClosed } r.participants[p.PeerID] = p r.attendance = append(r.attendance, &AttendanceRecord{ Username: username, PeerID: p.PeerID, JoinedAt: p.joinedAt, }) if len(r.attendance) > maxAttendance { r.attendance = r.attendance[len(r.attendance)-maxAttendance:] } r.lastActivity = time.Now() r.mu.Unlock() return p, nil } // RemoveParticipant unregisters a participant and closes its send channel. func (r *Room) RemoveParticipant(peerID int) { r.mu.Lock() _, ok := r.participants[peerID] if ok { delete(r.participants, peerID) } for _, record := range r.attendance { if record.PeerID == peerID && record.Present() { record.LeftAt = time.Now() break } } r.lastActivity = time.Now() r.mu.Unlock() if ok { r.channel.Leave(peerID) } } // KickParticipant removes a participant on the host's behalf, atomically // looking it up and unregistering it. It returns the removed participant so // the transport layer can announce the removal, and reports whether a // participant with that peer ID was present. Kicking marks the attendance // record as left, exactly like an ordinary leave. The host cannot kick // themselves: a request targeting the room host is refused. func (r *Room) KickParticipant(peerID int) (*Participant, bool) { r.mu.Lock() p, ok := r.participants[peerID] if !ok || p.IsHost { r.mu.Unlock() return nil, false } r.mu.Unlock() r.RemoveParticipant(peerID) return p, true } // Attendance returns a snapshot of the room's join / leave log in // chronological join order. func (r *Room) Attendance() []AttendanceRecord { r.mu.Lock() defer r.mu.Unlock() list := make([]AttendanceRecord, 0, len(r.attendance)) for _, record := range r.attendance { list = append(list, *record) } return list } // HasParticipantUsername reports whether a user with the given username is // currently connected to the room. func (r *Room) HasParticipantUsername(username string) bool { r.mu.Lock() defer r.mu.Unlock() for _, p := range r.participants { if p.Username == username { return true } } return false } // Participants returns a snapshot of the current participants. func (r *Room) Participants() []*Participant { r.mu.Lock() defer r.mu.Unlock() list := make([]*Participant, 0, len(r.participants)) for _, p := range r.participants { list = append(list, p) } return list } // ParticipantCount returns the number of connected participants. func (r *Room) ParticipantCount() int { r.mu.Lock() defer r.mu.Unlock() return len(r.participants) } // GetParticipant returns the participant with the given peer ID. func (r *Room) GetParticipant(peerID int) (*Participant, bool) { r.mu.Lock() defer r.mu.Unlock() p, ok := r.participants[peerID] return p, ok } // Broadcast queues msg to every channel subscriber except excludePeerID // (pass a negative value to send to everyone). For space-bound rooms this // includes generic space subscribers, so clients connected through the // sharedspace WebSocket receive meeting frames too. Full send buffers drop // the frame rather than blocking the room. func (r *Room) Broadcast(msg []byte, excludePeerID int) { r.channel.Broadcast(msg, excludePeerID) } // SendTo queues msg to a single channel subscriber. It reports whether the // peer exists on the room's channel. func (r *Room) SendTo(peerID int, msg []byte) bool { return r.channel.SendTo(peerID, msg) } // Touch refreshes the room's idle timer. func (r *Room) Touch() { r.mu.Lock() r.lastActivity = time.Now() r.mu.Unlock() } // SaveAttachment streams src to disk (up to maxSize bytes) and registers the // file in the room under a random ID. name is display-only and never used as // a filesystem path. Rooms with a bound shared space store the file in the // space instead, so AGI scripts can access it too. func (m *Manager) SaveAttachment(roomID string, name string, uploader string, src io.Reader, maxSize int64) (*Attachment, error) { room, ok := m.GetRoom(roomID) if !ok { return nil, ErrRoomNotFound } if maxSize <= 0 { maxSize = DefaultMaxUpload } //Space-backed room: the shared space owns the blob storage if space, ok := m.spaceOf(room); ok { itemType := sharedspace.ItemTypeFile if sharedspace.IsImageName(name) { itemType = sharedspace.ItemTypeImage } item, err := space.SaveBlob(itemType, name, uploader, OriginMeetRoom, src, maxSize) if err != nil { if err == sharedspace.ErrItemTooLarge { return nil, ErrAttachmentTooLarge } if err == sharedspace.ErrSpaceClosed { return nil, ErrRoomClosed } return nil, err } room.Touch() return &Attachment{ ID: item.ID, Name: item.Name, Size: item.Size, Uploader: item.Uploader, DiskPath: item.DiskPath, }, nil } idBytes := make([]byte, 16) if _, err := rand.Read(idBytes); err != nil { return nil, err } fileID := hex.EncodeToString(idBytes) roomDir := filepath.Join(m.storageRoot, roomID) if err := os.MkdirAll(roomDir, 0755); err != nil { return nil, err } diskPath := filepath.Join(roomDir, fileID) dst, err := os.Create(diskPath) if err != nil { return nil, err } written, err := io.Copy(dst, io.LimitReader(src, maxSize+1)) dst.Close() if err != nil { os.Remove(diskPath) return nil, err } if written > maxSize { os.Remove(diskPath) return nil, ErrAttachmentTooLarge } attachment := &Attachment{ ID: fileID, Name: clipString(name, maxNameLength), Size: written, Uploader: uploader, DiskPath: diskPath, } room.mu.Lock() if room.closed { room.mu.Unlock() os.Remove(diskPath) return nil, ErrRoomClosed } room.attachments[fileID] = attachment room.lastActivity = time.Now() room.mu.Unlock() return attachment, nil } // GetAttachment looks up a shared file in a room by its ID, consulting the // bound shared space when the room has one (covering both room uploads and // files posted into the space by AGI scripts). func (m *Manager) GetAttachment(roomID string, fileID string) (*Attachment, bool) { room, ok := m.GetRoom(roomID) if !ok { return nil, false } room.mu.Lock() attachment, ok := room.attachments[fileID] room.mu.Unlock() if ok { return attachment, true } if space, hasSpace := m.spaceOf(room); hasSpace { if item, found := space.GetItem(fileID); found && item.DiskPath != "" { return &Attachment{ ID: item.ID, Name: item.Name, Size: item.Size, Uploader: item.Uploader, DiskPath: item.DiskPath, }, true } } return nil, false } // LogChat mirrors a chat message into the room's bound shared space so AGI // scripts can read the meeting conversation. No-op for rooms without a space. func (m *Manager) LogChat(roomID string, username string, text string) { room, ok := m.GetRoom(roomID) if !ok { return } if space, hasSpace := m.spaceOf(room); hasSpace { space.AddText(username, text, OriginMeetRoom) } } // ListRoomsByHost returns a snapshot of the live rooms hosted by host. func (m *Manager) ListRoomsByHost(host string) []*Room { m.mu.RLock() defer m.mu.RUnlock() hosted := []*Room{} for _, room := range m.rooms { if room.Host == host { hosted = append(hosted, room) } } return hosted } // CloseRoom removes the room, closes every participant's send channel and // deletes the room's attachment files. Safe to call on an unknown ID. // It returns the removed room's participants so the transport layer can // finish delivering any queued frames before the sockets drop. func (m *Manager) CloseRoom(id string) []*Participant { m.mu.Lock() room, exists := m.rooms[id] if exists { delete(m.rooms, id) } m.mu.Unlock() if !exists { return nil } room.mu.Lock() room.closed = true members := make([]*Participant, 0, len(room.participants)) for _, p := range room.participants { members = append(members, p) } room.participants = make(map[int]*Participant) room.attachments = make(map[string]*Attachment) room.mu.Unlock() //Tearing the channel down closes every subscriber's send buffer (the //meeting participants and, for bound rooms, generic space subscribers) room.channel.Close() os.RemoveAll(filepath.Join(m.storageRoot, id)) //The bound shared space lives and dies with the room m.mu.RLock() sm := m.spaces m.mu.RUnlock() if sm != nil && room.SpaceID != "" { sm.DeleteSpace(room.SpaceID) } return members } // SweepIdleRooms closes every room that has no participants and has been // idle for longer than emptyIdle, returning the IDs it closed. func (m *Manager) SweepIdleRooms(emptyIdle time.Duration) []string { if emptyIdle <= 0 { emptyIdle = DefaultEmptyIdle } var toClose []string m.mu.RLock() for id, room := range m.rooms { room.mu.Lock() if len(room.participants) == 0 && time.Since(room.lastActivity) > emptyIdle { toClose = append(toClose, id) } room.mu.Unlock() } m.mu.RUnlock() for _, id := range toClose { m.CloseRoom(id) } return toClose } // RoomCount returns the number of live rooms. func (m *Manager) RoomCount() int { m.mu.RLock() defer m.mu.RUnlock() return len(m.rooms) } // FormatRoomID renders a room ID in the display form xxx-xxx-xxx. func FormatRoomID(id string) string { if len(id) != roomIDLength { return id } return fmt.Sprintf("%s-%s-%s", id[0:3], id[3:6], id[6:9]) } // NormalizeRoomID strips the separators FormatRoomID (or a user) may have // added, accepting inputs like "123-456-789" or "123 456 789". func NormalizeRoomID(id string) string { out := make([]rune, 0, len(id)) for _, c := range id { if c >= '0' && c <= '9' { out = append(out, c) } } return string(out) }