galene/rtpconn/webclient.go

package rtpconn

import (
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"os"
	"sync"
	"time"

	"github.com/gorilla/websocket"
	"github.com/pion/webrtc/v3"

	"github.com/jech/galene/conn"
	"github.com/jech/galene/diskwriter"
	"github.com/jech/galene/estimator"
	"github.com/jech/galene/group"
	"github.com/jech/galene/ice"
)

func errorToWSCloseMessage(id string, err error) (*clientMessage, []byte) {
	var code int
	var m *clientMessage
	var text string
	switch e := err.(type) {
	case *websocket.CloseError:
		code = websocket.CloseNormalClosure
	case group.ProtocolError:
		code = websocket.CloseProtocolError
		m = &clientMessage{
			Type:       "usermessage",
			Kind:       "error",
			Dest:       id,
			Privileged: true,
			Value:      e.Error(),
		}
		text = e.Error()
	case group.UserError, group.KickError:
		code = websocket.CloseNormalClosure
		m = errorMessage(id, err)
		text = e.Error()
	default:
		code = websocket.CloseInternalServerErr
	}
	return m, websocket.FormatCloseMessage(code, text)
}

func isWSNormalError(err error) bool {
	return websocket.IsCloseError(err,
		websocket.CloseNormalClosure,
		websocket.CloseGoingAway)
}

type webClient struct {
	group       *group.Group
	id          string
	username    string
	permissions []string
	data        map[string]interface{}
	requested   map[string][]string
	done        chan struct{}
	writeCh     chan interface{}
	writerDone  chan struct{}
	actionCh    chan struct{}

	mu   sync.Mutex
	down map[string]*rtpDownConnection
	up   map[string]*rtpUpConnection

	// action may be called with the group mutex taken, and therefore
	// actions needs to use its own mutex.
	actionMu sync.Mutex
	actions  []interface{}
}

func (c *webClient) Group() *group.Group {
	return c.group
}

func (c *webClient) Id() string {
	return c.id
}

func (c *webClient) Username() string {
	return c.username
}

func (c *webClient) SetUsername(username string) {
	c.username = username
}

func (c *webClient) Permissions() []string {
	return c.permissions
}

func (c *webClient) Data() map[string]interface{} {
	return c.data
}

func (c *webClient) SetPermissions(perms []string) {
	c.permissions = perms
}

func (c *webClient) PushClient(group, kind, id, username string, perms []string, data map[string]interface{}) error {
	return c.action(pushClientAction{
		group, kind, id, username, perms, data,
	})
}

type clientMessage struct {
	Type             string                   `json:"type"`
	Version          []string                 `json:"version"`
	Kind             string                   `json:"kind,omitempty"`
	Id               string                   `json:"id,omitempty"`
	Replace          string                   `json:"replace,omitempty"`
	Source           string                   `json:"source,omitempty"`
	Dest             string                   `json:"dest,omitempty"`
	Username         string                   `json:"username,omitempty"`
	Password         string                   `json:"password,omitempty"`
	Token            string                   `json:"token,omitempty"`
	Privileged       bool                     `json:"privileged,omitempty"`
	Permissions      []string                 `json:"permissions,omitempty"`
	Status           *group.Status            `json:"status,omitempty"`
	Data             map[string]interface{}   `json:"data,omitempty"`
	Group            string                   `json:"group,omitempty"`
	Value            interface{}              `json:"value,omitempty"`
	NoEcho           bool                     `json:"noecho,omitempty"`
	Time             int64                    `json:"time,omitempty"`
	SDP              string                   `json:"sdp,omitempty"`
	Candidate        *webrtc.ICECandidateInit `json:"candidate,omitempty"`
	Label            string                   `json:"label,omitempty"`
	Request          interface{}              `json:"request,omitempty"`
	RTCConfiguration *webrtc.Configuration    `json:"rtcConfiguration,omitempty"`
}

type closeMessage struct {
	data []byte
}

func getUpConn(c *webClient, id string) *rtpUpConnection {
	c.mu.Lock()
	defer c.mu.Unlock()

	if c.up == nil {
		return nil
	}
	return c.up[id]
}

func getUpConns(c *webClient) []*rtpUpConnection {
	c.mu.Lock()
	defer c.mu.Unlock()
	up := make([]*rtpUpConnection, 0, len(c.up))
	for _, u := range c.up {
		up = append(up, u)
	}
	return up
}

func addUpConn(c *webClient, id, label string, offer string) (*rtpUpConnection, bool, error) {
	c.mu.Lock()
	defer c.mu.Unlock()

	if c.up == nil {
		c.up = make(map[string]*rtpUpConnection)
	}
	if c.down != nil && c.down[id] != nil {
		return nil, false, errors.New("adding duplicate connection")
	}

	old := c.up[id]
	if old != nil {
		return old, false, nil
	}

	conn, err := newUpConn(c, id, label, offer)
	if err != nil {
		return nil, false, err
	}

	c.up[id] = conn

	conn.pc.OnICECandidate(func(candidate *webrtc.ICECandidate) {
		sendICE(c, id, candidate)
	})

	conn.pc.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) {
		if state == webrtc.ICEConnectionStateFailed {
			c.action(connectionFailedAction{id: id})
		}
	})

	return conn, true, nil
}

var ErrUserMismatch = errors.New("user id mismatch")

// delUpConn deletes an up connection.  If push is closed, the close is
// pushed to all corresponding down connections.
func delUpConn(c *webClient, id string, userId string, push bool) error {
	c.mu.Lock()
	if c.up == nil {
		c.mu.Unlock()
		return os.ErrNotExist
	}
	conn := c.up[id]
	if conn == nil {
		c.mu.Unlock()
		return os.ErrNotExist
	}
	if userId != "" {
		id, _ := conn.User()
		if id != userId {
			c.mu.Unlock()
			return ErrUserMismatch
		}
	}

	replace := conn.getReplace(false)

	delete(c.up, id)
	g := c.group
	c.mu.Unlock()

	conn.mu.Lock()
	conn.closed = true
	conn.mu.Unlock()

	conn.pc.Close()

	if push && g != nil {
		for _, c := range g.GetClients(c) {
			err := c.PushConn(g, id, nil, nil, replace)
			if err != nil {
				log.Printf("PushConn: %v", err)
			}
		}
	}

	return nil
}

func getDownConn(c *webClient, id string) *rtpDownConnection {
	if c.down == nil {
		return nil
	}

	c.mu.Lock()
	defer c.mu.Unlock()
	conn := c.down[id]
	if conn == nil {
		return nil
	}
	return conn
}

func getConn(c *webClient, id string) iceConnection {
	up := getUpConn(c, id)
	if up != nil {
		return up
	}
	down := getDownConn(c, id)
	if down != nil {
		return down
	}
	return nil
}

func addDownConn(c *webClient, remote conn.Up) (*rtpDownConnection, bool, error) {
	id := remote.Id()

	c.mu.Lock()
	defer c.mu.Unlock()

	if c.up != nil && c.up[id] != nil {
		return nil, false, errors.New("adding duplicate connection")
	}

	if c.down == nil {
		c.down = make(map[string]*rtpDownConnection)
	}

	if down := c.down[id]; down != nil {
		return down, false, nil
	}

	down, err := newDownConn(c, id, remote)
	if err != nil {
		return nil, false, err
	}

	down.pc.OnICECandidate(func(candidate *webrtc.ICECandidate) {
		sendICE(c, down.id, candidate)
	})

	down.pc.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) {
		if state == webrtc.ICEConnectionStateFailed {
			c.action(connectionFailedAction{id: down.id})
		}
	})

	err = remote.AddLocal(down)
	if err != nil {
		down.pc.Close()
		return nil, false, err
	}

	c.down[down.id] = down

	go rtcpDownSender(down)

	return down, true, nil
}

func delDownConn(c *webClient, id string) error {
	conn := delDownConnHelper(c, id)
	if conn != nil {
		conn.pc.Close()
		return nil
	}
	return os.ErrNotExist
}

func delDownConnHelper(c *webClient, id string) *rtpDownConnection {
	c.mu.Lock()
	defer c.mu.Unlock()

	if c.down == nil {
		return nil
	}
	conn := c.down[id]
	if conn == nil {
		return nil
	}

	conn.remote.DelLocal(conn)
	for _, track := range conn.tracks {
		// we only insert the track after we get an answer, so
		// ignore errors here.
		track.remote.DelLocal(track)
	}
	delete(c.down, id)
	return conn
}

var errUnexpectedTrackType = errors.New("unexpected track type, this shouldn't happen")

func addDownTrackUnlocked(conn *rtpDownConnection, remoteTrack *rtpUpTrack) error {
	for _, t := range conn.tracks {
		tt, ok := t.remote.(*rtpUpTrack)
		if !ok {
			return errUnexpectedTrackType
		}
		if tt == remoteTrack {
			return os.ErrExist
		}
	}

	id := remoteTrack.track.ID()
	if id == "" {
		log.Println("Got track with empty id")
		id = remoteTrack.track.RID()
	}
	if id == "" {
		id = remoteTrack.track.Kind().String()
	}
	msid := remoteTrack.track.StreamID()
	if msid == "" || msid == "-" {
		log.Println("Got track with empty msid")
		msid = remoteTrack.conn.Label()
	}
	if msid == "" {
		msid = "dummy"
	}

	local, err := webrtc.NewTrackLocalStaticRTP(
		remoteTrack.Codec(), id, msid,
	)
	if err != nil {
		return err
	}

	transceiver, err := conn.pc.AddTransceiverFromTrack(local,
		webrtc.RTPTransceiverInit{
			Direction: webrtc.RTPTransceiverDirectionSendonly,
		},
	)
	if err != nil {
		return err
	}

	codec := local.Codec()
	ptype, err := group.CodecPayloadType(local.Codec())
	if err != nil {
		log.Printf("Couldn't determine ptype for codec %v: %v",
			codec.MimeType, err)
	} else {
		err := transceiver.SetCodecPreferences(
			[]webrtc.RTPCodecParameters{
				{
					RTPCodecCapability: codec,
					PayloadType:        ptype,
				},
			},
		)
		if err != nil {
			log.Printf("Couldn't set ptype for codec %v: %v",
				codec.MimeType, err)
		}
	}

	parms := transceiver.Sender().GetParameters()
	if len(parms.Encodings) != 1 {
		return errors.New("got multiple encodings")
	}

	track := &rtpDownTrack{
		track:          local,
		sender:         transceiver.Sender(),
		ssrc:           parms.Encodings[0].SSRC,
		conn:           conn,
		remote:         remoteTrack,
		maxBitrate:     new(bitrate),
		maxREMBBitrate: new(bitrate),
		stats:          new(receiverStats),
		rate:           estimator.New(time.Second),
		atomics:        &downTrackAtomics{},
	}

	conn.tracks = append(conn.tracks, track)

	go rtcpDownListener(track)

	return nil
}

func delDownTrackUnlocked(conn *rtpDownConnection, track *rtpDownTrack) error {
	for i := range conn.tracks {
		if conn.tracks[i] == track {
			track.remote.DelLocal(track)
			conn.tracks =
				append(conn.tracks[:i], conn.tracks[i+1:]...)
			return conn.pc.RemoveTrack(track.sender)
		}
	}
	return os.ErrNotExist
}

func replaceTracks(conn *rtpDownConnection, remote []conn.UpTrack, limitSid bool) (bool, error) {
	conn.mu.Lock()
	defer conn.mu.Unlock()

	var add []*rtpUpTrack
	var del []*rtpDownTrack

outer:
	for _, rtrack := range remote {
		rt, ok := rtrack.(*rtpUpTrack)
		if !ok {
			return false, errUnexpectedTrackType
		}
		for _, track := range conn.tracks {
			rt2, ok := track.remote.(*rtpUpTrack)
			if !ok {
				return false, errUnexpectedTrackType
			}
			if rt == rt2 {
				continue outer
			}
		}
		add = append(add, rt)
	}

outer2:
	for _, track := range conn.tracks {
		rt, ok := track.remote.(*rtpUpTrack)
		if !ok {
			return false, errUnexpectedTrackType
		}
		for _, rtrack := range remote {
			rt2, ok := rtrack.(*rtpUpTrack)
			if !ok {
				return false, errUnexpectedTrackType
			}
			if rt == rt2 {
				continue outer2
			}
		}
		del = append(del, track)
	}

	defer func() {
		for _, t := range conn.tracks {
			layer := t.getLayerInfo()
			layer.limitSid = limitSid
			if limitSid {
				layer.wantedSid = 0
			}
			t.setLayerInfo(layer)
		}
	}()

	if len(del) == 0 && len(add) == 0 {
		return false, nil
	}

	for _, t := range del {
		err := delDownTrackUnlocked(conn, t)
		if err != nil {
			return false, err
		}
	}

	for _, rt := range add {
		err := addDownTrackUnlocked(conn, rt)
		if err != nil {
			return false, err
		}
	}

	return true, nil
}

func negotiate(c *webClient, down *rtpDownConnection, restartIce bool, replace string) error {
	if down.pc.SignalingState() == webrtc.SignalingStateHaveLocalOffer {
		// avoid sending multiple offers back-to-back
		if restartIce {
			down.negotiationNeeded = negotiationRestartIce
		} else if down.negotiationNeeded == negotiationUnneeded {
			down.negotiationNeeded = negotiationNeeded
		}
		return nil
	}

	down.negotiationNeeded = negotiationUnneeded

	options := webrtc.OfferOptions{ICERestart: restartIce}
	offer, err := down.pc.CreateOffer(&options)
	if err != nil {
		return err
	}

	err = down.pc.SetLocalDescription(offer)
	if err != nil {
		return err
	}

	source, username := down.remote.User()

	return c.write(clientMessage{
		Type:     "offer",
		Id:       down.id,
		Label:    down.remote.Label(),
		Replace:  replace,
		Source:   source,
		Username: username,
		SDP:      down.pc.LocalDescription().SDP,
	})
}

func sendICE(c *webClient, id string, candidate *webrtc.ICECandidate) error {
	if candidate == nil {
		return nil
	}
	cand := candidate.ToJSON()
	return c.write(clientMessage{
		Type:      "ice",
		Id:        id,
		Candidate: &cand,
	})
}

func gotOffer(c *webClient, id, label string, sdp string, replace string) error {
	up, _, err := addUpConn(c, id, label, sdp)
	if err != nil {
		return err
	}

	if replace != "" {
		up.replace = replace
		delUpConn(c, replace, c.Id(), false)
	}

	err = up.pc.SetRemoteDescription(webrtc.SessionDescription{
		Type: webrtc.SDPTypeOffer,
		SDP:  sdp,
	})
	if err != nil {
		return err
	}

	answer, err := up.pc.CreateAnswer(nil)
	if err != nil {
		return err
	}

	err = up.pc.SetLocalDescription(answer)
	if err != nil {
		return err
	}

	err = up.flushICECandidates()
	if err != nil {
		log.Printf("ICE: %v", err)
	}

	return c.write(clientMessage{
		Type: "answer",
		Id:   id,
		SDP:  up.pc.LocalDescription().SDP,
	})
}

var ErrUnknownId = errors.New("unknown id")

func gotAnswer(c *webClient, id string, sdp string) error {
	down := getDownConn(c, id)
	if down == nil {
		return ErrUnknownId
	}

	err := down.pc.SetRemoteDescription(webrtc.SessionDescription{
		Type: webrtc.SDPTypeAnswer,
		SDP:  sdp,
	})
	if err != nil {
		return err
	}

	err = down.flushICECandidates()
	if err != nil {
		log.Printf("ICE: %v", err)
	}

	add := func() {
		down.pc.OnConnectionStateChange(nil)
		for _, t := range down.tracks {
			err := t.remote.AddLocal(t)
			if err != nil && err != os.ErrClosed {
				log.Printf("Add track: %v", err)
			}
		}
	}
	down.pc.OnConnectionStateChange(func(state webrtc.PeerConnectionState) {
		if state == webrtc.PeerConnectionStateConnected {
			add()
		}
	})
	if down.pc.ConnectionState() == webrtc.PeerConnectionStateConnected {
		add()
	}

	return nil
}

func gotICE(c *webClient, candidate *webrtc.ICECandidateInit, id string) error {
	conn := getConn(c, id)
	if conn == nil {
		return errors.New("unknown id in ICE")
	}
	return conn.addICECandidate(candidate)
}

var errBadType = errors.New("bad type")

func toStringArray(r interface{}) ([]string, error) {
	if r == nil {
		return nil, nil
	}
	rr, ok := r.([]interface{})
	if !ok {
		return nil, errBadType
	}
	if rr == nil {
		return nil, nil
	}

	rrr := make([]string, len(rr))
	for i, s := range rr {
		rrr[i], ok = s.(string)
		if !ok {
			return nil, errBadType
		}
	}
	return rrr, nil
}

func parseRequested(r interface{}) (map[string][]string, error) {
	if r == nil {
		return nil, nil
	}
	rr, ok := r.(map[string]interface{})
	if !ok {
		return nil, errBadType
	}
	if rr == nil {
		return nil, nil
	}
	rrr := make(map[string][]string)
	for k, v := range rr {
		vv, err := toStringArray(v)
		if err != nil {
			return nil, err
		}
		rrr[k] = vv
	}
	return rrr, nil
}

func (c *webClient) setRequested(requested map[string][]string) error {
	if c.group == nil {
		return errors.New("attempted to request with no group joined")
	}
	c.requested = requested

	requestConns(c, c.group, "")
	return nil
}

func (c *webClient) setRequestedStream(down *rtpDownConnection, requested []string) error {
	var remoteClient group.Client
	remote, ok := down.remote.(*rtpUpConnection)
	if ok {
		remoteClient = remote.client
	}
	down.requested = requested
	return remoteClient.RequestConns(c, c.group, remote.id)
}

func (c *webClient) RequestConns(target group.Client, g *group.Group, id string) error {
	return c.action(requestConnsAction{g, target, id})
}

func requestConns(target group.Client, g *group.Group, id string) {
	clients := g.GetClients(target)
	for _, c := range clients {
		c.RequestConns(target, g, id)
	}
}

func requestedTracks(c *webClient, override []string, up conn.Up, tracks []conn.UpTrack) ([]conn.UpTrack, bool) {
	r := override
	if r == nil {
		var ok bool
		r, ok = c.requested[up.Label()]
		if !ok {
			r, ok = c.requested[""]
		}
		if !ok || len(r) == 0 {
			return nil, false
		}
	}

	var audio, video, videoLow bool
	for _, s := range r {
		switch s {
		case "audio":
			audio = true
		case "video":
			video = true
		case "video-low":
			videoLow = true
		default:
			log.Printf("client requested unknown value %v", s)
		}
	}

	find := func(kind webrtc.RTPCodecType, labels ...string) conn.UpTrack {
		for _, l := range labels {
			for _, t := range tracks {
				if t.Kind() != kind {
					continue
				}
				if t.Label() == l {
					return t
				}
			}
		}
		for _, t := range tracks {
			if t.Kind() != kind {
				continue
			}
			return t
		}
		return nil
	}

	var ts []conn.UpTrack
	limitSid := false
	if audio {
		t := find(webrtc.RTPCodecTypeAudio)
		if t != nil {
			ts = append(ts, t)
		}
	}
	if video {
		t := find(
			webrtc.RTPCodecTypeVideo, "h", "m", "video",
		)
		if t != nil {
			ts = append(ts, t)
		}
	} else if videoLow {
		t := find(
			webrtc.RTPCodecTypeVideo, "l", "m", "video",
		)
		if t != nil {
			ts = append(ts, t)
			if t.Label() != "l" {
				limitSid = true
			}
		}
	}

	return ts, limitSid
}

func (c *webClient) PushConn(g *group.Group, id string, up conn.Up, tracks []conn.UpTrack, replace string) error {
	err := c.action(pushConnAction{g, id, up, tracks, replace})
	if err != nil {
		return err
	}
	return nil
}

func readMessage(conn *websocket.Conn, m *clientMessage) error {
	err := conn.SetReadDeadline(time.Now().Add(15 * time.Second))
	if err != nil {
		return err
	}
	defer conn.SetReadDeadline(time.Time{})

	return conn.ReadJSON(&m)
}

func StartClient(conn *websocket.Conn) (err error) {
	var m clientMessage

	err = readMessage(conn, &m)
	if err != nil {
		conn.Close()
		return
	}

	if m.Type != "handshake" {
		conn.WriteMessage(websocket.CloseMessage,
			websocket.FormatCloseMessage(
				websocket.CloseProtocolError,
				"you must handshake first",
			),
		)
		conn.Close()
		err = group.ProtocolError("client didn't handshake")
		return
	}

	versionError := true
	if m.Version != nil {
		for _, v := range m.Version {
			if v == "1" {
				versionError = false
			}
		}
	}

	c := &webClient{
		id:       m.Id,
		actionCh: make(chan struct{}, 1),
		done:     make(chan struct{}),
	}

	defer close(c.done)

	c.writeCh = make(chan interface{}, 100)
	c.writerDone = make(chan struct{})
	go clientWriter(conn, c.writeCh, c.writerDone)
	defer func() {
		m, e := errorToWSCloseMessage(c.id, err)
		if isWSNormalError(err) {
			err = nil
		} else if _, ok := err.(group.KickError); ok {
			err = nil
		}
		if m != nil {
			c.write(*m)
		}
		c.close(e)
	}()

	return clientLoop(c, conn, versionError)
}

type pushConnAction struct {
	group   *group.Group
	id      string
	conn    conn.Up
	tracks  []conn.UpTrack
	replace string
}

type requestConnsAction struct {
	group  *group.Group
	target group.Client
	id     string
}

type connectionFailedAction struct {
	id string
}

type pushClientAction struct {
	group       string
	kind        string
	id          string
	username    string
	permissions []string
	data        map[string]interface{}
}

type permissionsChangedAction struct{}

type joinedAction struct {
	group string
	kind  string
}

type kickAction struct {
	id       string
	username string
	message  string
}

var errEmptyId = group.ProtocolError("empty id")

func member(v string, l []string) bool {
	for _, w := range l {
		if v == w {
			return true
		}
	}
	return false
}

func remove(v string, l []string) []string {
	for i, w := range l {
		if v == w {
			l = append(l[:i], l[i+1:]...)
			return l
		}
	}
	return l
}

func addnew(v string, l []string) []string {
	if member(v, l) {
		return l
	}
	l = append(l, v)
	return l
}

func clientLoop(c *webClient, ws *websocket.Conn, versionError bool) error {
	read := make(chan interface{}, 1)
	go clientReader(ws, read, c.done)

	defer leaveGroup(c)

	readTime := time.Now()

	ticker := time.NewTicker(10 * time.Second)
	defer ticker.Stop()

	err := c.write(clientMessage{
		Type:    "handshake",
		Version: []string{"1"},
	})
	if err != nil {
		return err
	}

	if versionError {
		c.write(clientMessage{
			Type:       "usermessage",
			Kind:       "warning",
			Dest:       c.id,
			Privileged: true,
			Value: "This client is using an unknown protocol version.\n" +
				"Perhaps it needs upgrading?\n" +
				"Trying to continue, things may break.",
		})
	}

	for {
		select {
		case m, ok := <-read:
			if !ok {
				return errors.New("reader died")
			}
			switch m := m.(type) {
			case clientMessage:
				readTime = time.Now()
				err := handleClientMessage(c, m)
				if err != nil {
					return err
				}
			case error:
				return m
			}
		case <-c.actionCh:
			c.actionMu.Lock()
			actions := c.actions
			c.actions = nil
			c.actionMu.Unlock()
			for _, a := range actions {
				err := handleAction(c, a)
				if err != nil {
					return err
				}
			}
		case <-ticker.C:
			if time.Since(readTime) > 75*time.Second {
				return errors.New("client is dead")
			}
			// Some reverse proxies timeout connexions at 60
			// seconds, make sure we generate some activity
			// after 55s at most.
			if time.Since(readTime) > 45*time.Second {
				err := c.write(clientMessage{
					Type: "ping",
				})
				if err != nil {
					return err
				}
			}
		}
	}
}

func pushDownConn(c *webClient, id string, up conn.Up, tracks []conn.UpTrack, replace string) error {
	var requested []conn.UpTrack
	limitSid := false
	if up != nil {
		var old *rtpDownConnection
		if replace != "" {
			old = getDownConn(c, replace)
		} else {
			old = getDownConn(c, up.Id())
		}
		var override []string
		if old != nil {
			override = old.requested
		}
		requested, limitSid = requestedTracks(c, override, up, tracks)
	}

	if replace != "" {
		err := delDownConn(c, replace)
		if err != nil {
			log.Printf("Replace: %v", err)
		}
	}

	// closes over replace, which will be modified below
	defer func() {
		if replace != "" {
			closeDownConn(c, replace, "")
		}
	}()

	if len(requested) == 0 {
		closeDownConn(c, id, "")
		return nil
	}

	down, _, err := addDownConn(c, up)
	if err != nil {
		if err == os.ErrClosed {
			return nil
		}
		return err
	}
	done, err := replaceTracks(down, requested, limitSid)
	if err != nil || !done {
		return err
	}
	err = negotiate(c, down, false, replace)
	if err != nil {
		log.Printf("Negotiation failed: %v", err)
		closeDownConn(c, down.id, err.Error())
		return err
	}
	replace = ""
	return nil
}

func handleAction(c *webClient, a interface{}) error {
	switch a := a.(type) {
	case pushConnAction:
		if c.group == nil || c.group != a.group {
			log.Printf("Got connectsions for wrong group")
			return nil
		}
		return pushDownConn(c, a.id, a.conn, a.tracks, a.replace)
	case requestConnsAction:
		g := c.group
		if g == nil || a.group != g {
			log.Printf("Misdirected pushConns")
			return nil
		}
		for _, u := range c.up {
			if a.id != "" && a.id != u.id {
				continue
			}
			tracks := u.getTracks()
			replace := u.getReplace(false)

			ts := make([]conn.UpTrack, len(tracks))
			for i, t := range tracks {
				ts[i] = t
			}
			err := a.target.PushConn(g, u.id, u, ts, replace)
			if err != nil {
				log.Printf("PushConn: %v", err)
			}
		}
	case connectionFailedAction:
		if down := getDownConn(c, a.id); down != nil {
			err := negotiate(c, down, true, "")
			if err != nil {
				return err
			}
			tracks := make(
				[]conn.UpTrack, len(down.tracks),
			)
			for i, t := range down.tracks {
				tracks[i] = t.remote
			}
			c.PushConn(
				c.group,
				down.remote.Id(), down.remote,
				tracks, "",
			)
		} else if up := getUpConn(c, a.id); up != nil {
			c.write(clientMessage{
				Type: "renegotiate",
				Id:   a.id,
			})
		} else {
			log.Printf("Attempting to renegotiate " +
				"unknown connection")
		}

	case pushClientAction:
		if a.group != c.group.Name() {
			log.Printf("got client for wrong group")
			return nil
		}
		perms := append([]string(nil), a.permissions...)
		return c.write(clientMessage{
			Type:        "user",
			Kind:        a.kind,
			Id:          a.id,
			Username:    a.username,
			Permissions: perms,
			Data:        a.data,
		})
	case joinedAction:
		var status *group.Status
		var data map[string]interface{}
		if a.group != "" {
			g := group.Get(a.group)
			if g != nil {
				s := g.Status(true, "")
				status = &s
				data = g.Data()
			}
		}
		perms := append([]string(nil), c.permissions...)
		err := c.write(clientMessage{
			Type:             "joined",
			Kind:             a.kind,
			Group:            a.group,
			Username:         c.username,
			Permissions:      perms,
			Status:           status,
			Data:             data,
			RTCConfiguration: ice.ICEConfiguration(),
		})
		if err != nil {
			return err
		}
		h := c.group.GetChatHistory()
		for _, m := range h {
			err := c.write(clientMessage{
				Type:     "chathistory",
				Source:   m.Id,
				Username: m.User,
				Time:     m.Time,
				Value:    m.Value,
				Kind:     m.Kind,
			})
			if err != nil {
				return err
			}
		}
	case permissionsChangedAction:
		g := c.Group()
		if g == nil {
			return errors.New("Permissions changed in no group")
		}
		perms := append([]string(nil), c.permissions...)
		status := g.Status(true, "")
		c.write(clientMessage{
			Type:             "joined",
			Kind:             "change",
			Group:            g.Name(),
			Username:         c.username,
			Permissions:      perms,
			Status:           &status,
			RTCConfiguration: ice.ICEConfiguration(),
		})
		if !member("present", c.permissions) {
			up := getUpConns(c)
			for _, u := range up {
				err := delUpConn(
					c, u.id, c.id, true,
				)
				if err == nil {
					failUpConnection(
						c, u.id,
						"permission denied",
					)
				}
			}
		}
		id := c.Id()
		user := c.Username()
		d := c.Data()
		clients := g.GetClients(nil)
		go func(clients []group.Client) {
			for _, cc := range clients {
				cc.PushClient(
					g.Name(), "change", id, user, perms, d,
				)
			}
		}(clients)
	case kickAction:
		return group.KickError{
			a.id, a.username, a.message,
		}
	default:
		log.Printf("unexpected action %T", a)
		return errors.New("unexpected action")
	}
	return nil
}

func failUpConnection(c *webClient, id string, message string) error {
	if id != "" {
		err := c.write(clientMessage{
			Type: "abort",
			Id:   id,
		})
		if err != nil {
			return err
		}
	}
	if message != "" {
		err := c.error(group.UserError(message))
		if err != nil {
			return err
		}
	}
	return nil
}

func leaveGroup(c *webClient) {
	if c.group == nil {
		return
	}

	if c.up != nil {
		for id := range c.up {
			delUpConn(c, id, c.id, true)
		}
	}
	if c.down != nil {
		for id := range c.down {
			delDownConn(c, id)
		}
	}

	group.DelClient(c)
	c.permissions = nil
	c.data = nil
	c.requested = make(map[string][]string)
	c.group = nil
}

func closeDownConn(c *webClient, id string, message string) error {
	err := delDownConn(c, id)
	if err != nil && !os.IsNotExist(err) {
		log.Printf("Close down connection: %v", err)
	}
	err = c.write(clientMessage{
		Type: "close",
		Id:   id,
	})
	if err != nil {
		return err
	}
	if message != "" {
		err := c.error(group.UserError(message))
		if err != nil {
			return err
		}
	}
	return nil
}

func setPermissions(g *group.Group, id string, perm string) error {
	client := g.GetClient(id)
	if client == nil {
		return group.UserError("no such user")
	}

	c, ok := client.(*webClient)
	if !ok {
		return group.UserError("this is not a real user")
	}

	switch perm {
	case "op":
		c.permissions = addnew("op", c.permissions)
		if g.Description().AllowRecording {
			c.permissions = addnew("record", c.permissions)
		}
	case "unop":
		c.permissions = remove("op", c.permissions)
		c.permissions = remove("record", c.permissions)
	case "present":
		c.permissions = addnew("present", c.permissions)
	case "unpresent":
		c.permissions = remove("present", c.permissions)
	default:
		return group.UserError("unknown permission")
	}
	return c.action(permissionsChangedAction{})
}

func (c *webClient) Kick(id, user, message string) error {
	return c.action(kickAction{id, user, message})
}

func (c *webClient) Joined(group, kind string) error {
	return c.action(joinedAction{group, kind})
}

func kickClient(g *group.Group, id, user, dest string, message string) error {
	client := g.GetClient(dest)
	if client == nil {
		return group.UserError("no such user")
	}

	return client.Kick(id, user, message)
}

func handleClientMessage(c *webClient, m clientMessage) error {
	if m.Source != "" {
		if m.Source != c.Id() {
			return group.ProtocolError("spoofed client id")
		}
	}

	if m.Type != "join" {
		if m.Username != "" {
			if m.Username != c.Username() {
				return group.ProtocolError("spoofed username")
			}
		}
	}

	switch m.Type {
	case "join":
		if m.Kind == "leave" {
			if c.group == nil || c.group.Name() != m.Group {
				return group.ProtocolError("you are not joined")
			}
			leaveGroup(c)
		}

		if m.Kind != "join" {
			return group.ProtocolError("unknown kind")
		}

		if c.group != nil {
			return group.ProtocolError(
				"cannot join multiple groups",
			)
		}
		c.username = m.Username
		c.data = m.Data
		g, err := group.AddClient(m.Group, c,
			group.ClientCredentials{
				Username: m.Username,
				Password: m.Password,
				Token:    m.Token,
			},
		)
		if err != nil {
			var s string
			if os.IsNotExist(err) {
				s = "group does not exist"
			} else if err == group.ErrNotAuthorised {
				s = "not authorised"
				time.Sleep(200 * time.Millisecond)
			} else if err == group.ErrAnonymousNotAuthorised {
				s = "please choose a username"
			} else if e, ok := err.(group.UserError); ok {
				s = string(e)
			} else {
				s = "internal server error"
				log.Printf("Join group: %v", err)
			}
			return c.write(clientMessage{
				Type:     "joined",
				Kind:     "fail",
				Group:    m.Group,
				Username: c.username,
				Value:    s,
			})
		}
		if redirect := g.Description().Redirect; redirect != "" {
			// We normally redirect at the HTTP level, but the group
			// description could have been edited in the meantime.
			return c.write(clientMessage{
				Type:     "joined",
				Kind:     "redirect",
				Group:    m.Group,
				Username: c.username,
				Value:    redirect,
			})
		}
		c.group = g
	case "request":
		requested, err := parseRequested(m.Request)
		if err != nil {
			return err
		}
		return c.setRequested(requested)
	case "requestStream":
		down := getDownConn(c, m.Id)
		if down == nil {
			return ErrUnknownId
		}
		requested, err := toStringArray(m.Request)
		if err != nil {
			return err
		}
		c.setRequestedStream(down, requested)
	case "offer":
		if m.Id == "" {
			return errEmptyId
		}
		if !member("present", c.permissions) {
			if m.Replace != "" {
				delUpConn(c, m.Replace, c.id, true)
			}
			c.write(clientMessage{
				Type: "abort",
				Id:   m.Id,
			})
			return c.error(group.UserError("not authorised"))
		}
		err := gotOffer(c, m.Id, m.Label, m.SDP, m.Replace)
		if err != nil {
			log.Printf("gotOffer: %v", err)
			return failUpConnection(c, m.Id, err.Error())
		}
	case "answer":
		if m.Id == "" {
			return errEmptyId
		}
		err := gotAnswer(c, m.Id, m.SDP)
		if err != nil {
			log.Printf("gotAnswer: %v", err)
			message := ""
			if err != ErrUnknownId {
				message = err.Error()
			}
			return closeDownConn(c, m.Id, message)
		}
		down := getDownConn(c, m.Id)
		if down == nil {
			return ErrUnknownId
		}
		if down.negotiationNeeded > negotiationUnneeded {
			err := negotiate(
				c, down,
				down.negotiationNeeded == negotiationRestartIce,
				"",
			)
			if err != nil {
				return closeDownConn(c, m.Id, err.Error())
			}
		}
	case "renegotiate":
		if m.Id == "" {
			return errEmptyId
		}
		down := getDownConn(c, m.Id)
		if down != nil {
			err := negotiate(c, down, true, "")
			if err != nil {
				return closeDownConn(c, m.Id, err.Error())
			}
		} else {
			log.Printf("Trying to renegotiate unknown connection")
		}
	case "close":
		if m.Id == "" {
			return errEmptyId
		}
		err := delUpConn(c, m.Id, c.id, true)
		if err != nil {
			log.Printf("Deleting up connection %v: %v",
				m.Id, err)
			return nil
		}
	case "abort":
		if m.Id == "" {
			return errEmptyId
		}
		return closeDownConn(c, m.Id, "")
	case "ice":
		if m.Id == "" {
			return errEmptyId
		}
		if m.Candidate == nil {
			return group.ProtocolError("null candidate")
		}
		err := gotICE(c, m.Candidate, m.Id)
		if err != nil {
			log.Printf("ICE: %v", err)
		}
	case "chat", "usermessage":
		g := c.group
		if g == nil {
			return c.error(group.UserError("join a group first"))
		}

		tm := group.ToJSTime(time.Now())

		if m.Type == "chat" {
			if m.Dest == "" {
				g.AddToChatHistory(
					m.Source, m.Username, tm, m.Kind, m.Value,
				)
			}
		}
		mm := clientMessage{
			Type:       m.Type,
			Source:     m.Source,
			Dest:       m.Dest,
			Username:   m.Username,
			Privileged: member("op", c.permissions),
			Time:       tm,
			Kind:       m.Kind,
			NoEcho:     m.NoEcho,
			Value:      m.Value,
		}
		if m.Dest == "" {
			var except group.Client
			if m.NoEcho {
				except = c
			}
			err := broadcast(g.GetClients(except), mm)
			if err != nil {
				log.Printf("broadcast(chat): %v", err)
			}
		} else {
			cc := g.GetClient(m.Dest)
			if cc == nil {
				return c.error(group.UserError("user unknown"))
			}
			ccc, ok := cc.(*webClient)
			if !ok {
				return c.error(group.UserError(
					"this user doesn't chat",
				))
			}
			ccc.write(mm)
		}
	case "groupaction":
		g := c.group
		if g == nil {
			return c.error(group.UserError("join a group first"))
		}
		switch m.Kind {
		case "clearchat":
			g.ClearChatHistory()
			m := clientMessage{
				Type:       "usermessage",
				Kind:       "clearchat",
				Privileged: true,
			}
			err := broadcast(g.GetClients(nil), m)
			if err != nil {
				log.Printf("broadcast(clearchat): %v", err)
			}
		case "lock", "unlock":
			if !member("op", c.permissions) {
				return c.error(group.UserError("not authorised"))
			}
			message := ""
			v, ok := m.Value.(string)
			if ok {
				message = v
			}
			g.SetLocked(m.Kind == "lock", message)
		case "record":
			if !member("record", c.permissions) {
				return c.error(group.UserError("not authorised"))
			}
			for _, cc := range g.GetClients(c) {
				_, ok := cc.(*diskwriter.Client)
				if ok {
					return c.error(group.UserError("already recording"))
				}
			}
			disk := diskwriter.New(g)
			_, err := group.AddClient(g.Name(), disk,
				group.ClientCredentials{
					System: true,
				},
			)
			if err != nil {
				disk.Close()
				return c.error(err)
			}
			requestConns(disk, c.group, "")
		case "unrecord":
			if !member("record", c.permissions) {
				return c.error(group.UserError("not authorised"))
			}
			for _, cc := range g.GetClients(c) {
				disk, ok := cc.(*diskwriter.Client)
				if ok {
					disk.Close()
					group.DelClient(disk)
				}
			}
		case "subgroups":
			if !member("op", c.permissions) {
				return c.error(group.UserError("not authorised"))
			}
			s := ""
			for _, sg := range group.GetSubGroups(g.Name()) {
				plural := ""
				if sg.Clients > 1 {
					plural = "s"
				}
				s = s + fmt.Sprintf("%v (%v client%v)\n",
					sg.Name, sg.Clients, plural)
			}
			c.write(clientMessage{
				Type:     "chat",
				Dest:     c.id,
				Username: "Server",
				Time:     group.ToJSTime(time.Now()),
				Value:    s,
			})
		case "setdata":
			if !member("op", c.permissions) {
				return c.error(group.UserError("not authorised"))
			}
			data, ok := m.Value.(map[string]interface{})
			if !ok {
				return c.error(group.UserError(
					"Bad value in setdata",
				))
			}
			g.UpdateData(data)
		default:
			return group.ProtocolError("unknown group action")
		}
	case "useraction":
		g := c.group
		if g == nil {
			return c.error(group.UserError("join a group first"))
		}
		switch m.Kind {
		case "op", "unop", "present", "unpresent":
			if !member("op", c.permissions) {
				return c.error(group.UserError("not authorised"))
			}
			err := setPermissions(g, m.Dest, m.Kind)
			if err != nil {
				return c.error(err)
			}
		case "kick":
			if !member("op", c.permissions) {
				return c.error(group.UserError("not authorised"))
			}
			message := ""
			v, ok := m.Value.(string)
			if ok {
				message = v
			}
			err := kickClient(g, m.Source, m.Username, m.Dest, message)
			if err != nil {
				return c.error(err)
			}
		case "setdata":
			if m.Dest != c.Id() {
				return c.error(group.UserError("not authorised"))
			}
			data, ok := m.Value.(map[string]interface{})
			if !ok {
				return c.error(group.UserError(
					"Bad value in setdata",
				))
			}
			if c.data == nil {
				c.data = make(map[string]interface{})
			}
			for k, v := range data {
				if v == nil {
					delete(c.data, k)
				} else {
					c.data[k] = v
				}
			}
			id := c.Id()
			user := c.Username()
			perms := c.Permissions()
			data = c.Data()
			go func(clients []group.Client) {
				for _, cc := range clients {
					cc.PushClient(
						g.Name(), "change",
						id, user, perms, data,
					)
				}
			}(g.GetClients(nil))
		default:
			return group.ProtocolError("unknown user action")
		}
	case "pong":
		// nothing
	case "ping":
		return c.write(clientMessage{
			Type: "pong",
		})
	default:
		log.Printf("unexpected message: %v", m.Type)
		return group.ProtocolError("unexpected message")
	}
	return nil
}

func clientReader(conn *websocket.Conn, read chan<- interface{}, done <-chan struct{}) {
	defer close(read)
	for {
		var m clientMessage
		err := conn.ReadJSON(&m)
		if err != nil {
			select {
			case read <- err:
				return
			case <-done:
				return
			}
		}
		select {
		case read <- m:
		case <-done:
			return
		}
	}
}

func clientWriter(conn *websocket.Conn, ch <-chan interface{}, done chan<- struct{}) {
	defer func() {
		close(done)
		conn.Close()
	}()

	for {
		m, ok := <-ch
		if !ok {
			break
		}
		err := conn.SetWriteDeadline(
			time.Now().Add(500 * time.Millisecond),
		)
		if err != nil {
			return
		}
		switch m := m.(type) {
		case clientMessage:
			err := conn.WriteJSON(m)
			if err != nil {
				return
			}
		case []byte:
			err := conn.WriteMessage(websocket.TextMessage, m)
			if err != nil {
				return
			}
		case closeMessage:
			if m.data != nil {
				conn.WriteMessage(
					websocket.CloseMessage,
					m.data,
				)
			}
			return
		default:
			log.Printf("clientWiter: unexpected message %T", m)
			return
		}
	}
}

func (c *webClient) Warn(oponly bool, message string) error {
	if oponly && !member("op", c.permissions) {
		return nil
	}

	return c.write(clientMessage{
		Type:       "usermessage",
		Kind:       "warning",
		Dest:       c.id,
		Privileged: true,
		Value:      message,
	})
}

var ErrClientDead = errors.New("client is dead")

func (c *webClient) action(a interface{}) error {
	c.actionMu.Lock()
	empty := len(c.actions) == 0
	c.actions = append(c.actions, a)
	c.actionMu.Unlock()

	if empty {
		select {
		case c.actionCh <- struct{}{}:
			return nil
		case <-c.done:
			return ErrClientDead
		default:
		}
	}
	return nil
}

func (c *webClient) write(m clientMessage) error {
	select {
	case c.writeCh <- m:
		return nil
	case <-c.writerDone:
		return ErrClientDead
	}
}

func broadcast(cs []group.Client, m clientMessage) error {
	b, err := json.Marshal(m)
	if err != nil {
		return err
	}
	for _, c := range cs {
		cc, ok := c.(*webClient)
		if !ok {
			continue
		}
		select {
		case cc.writeCh <- b:
		case <-cc.writerDone:
		}
	}
	return nil
}

func (c *webClient) close(data []byte) error {
	select {
	case c.writeCh <- closeMessage{data}:
		return nil
	case <-c.writerDone:
		return ErrClientDead
	}
}

func errorMessage(id string, err error) *clientMessage {
	switch e := err.(type) {
	case group.UserError:
		return &clientMessage{
			Type:       "usermessage",
			Kind:       "error",
			Dest:       id,
			Privileged: true,
			Value:      e.Error(),
		}
	case group.KickError:
		message := e.Message
		if message == "" {
			message = "you have been kicked out"
		}
		return &clientMessage{
			Type:       "usermessage",
			Kind:       "kicked",
			Id:         e.Id,
			Username:   e.Username,
			Dest:       id,
			Privileged: true,
			Value:      message,
		}
	default:
		return nil
	}
}

func (c *webClient) error(err error) error {
	m := errorMessage(c.id, err)
	if m == nil {
		return err
	}
	return c.write(*m)
}