package localserver

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"strconv"
	sync "sync"
	"sync/atomic"
	"time"

	"github.com/google/uuid"
	"github.com/pion/rtcp"
	"github.com/pion/webrtc/v3"
)

const (
	AUDIO_CHANNEL_ACCESS_DENIED                ReqType = "audio_channel_access_denied"
	AUDIO_CHANNEL_STOP_CALL                    ReqType = "audio_channel_stop_call"
	AUDIO_CHANNEL_ACCESS_GRANTED               ReqType = "audio_channel_access_granted"
	AUDIO_CHANNEL_WEBRTC_OFFER                 ReqType = "audio_channel_offer"
	AUDIO_CHANNEL_WEBRTC_ANSWER                ReqType = "audio_channel_answer"
	AUDIO_CHANNEL_WEBRTC_RENNEGOTIATION_OFFER  ReqType = "audio_channel_rennegotiation_offer"
	AUDIO_CHANNEL_WEBRTC_RENNEGOTIATION_ANSWER ReqType = "audio_channel_rennegotiation_answer"
	AUDIO_CHANNEL_WEBRTC_COUNTER_OFFER         ReqType = "audio_channel_webrtc_counter_offer"
	AUDIO_CHANNEL_WEBRTC_CANDIDATE             ReqType = "audio_channel_webrtc_candidate"
	AUDIO_CHANNEL_REMOVE_VIDEO                 ReqType = "audio_channel_remove_video"
	GET_AUDIO_CHANNEL_TRACKS                   ReqType = "audio_channel_get_tracks"
)

const (
	AUDIO_CHANNEL_USER_MUTE             = "audio_channel_user_mute"
	AUDIO_CHANNEL_USER_UNMUTE           = "audio_channel_user_unmute"
	AUDIO_CHANNEL_USER_SPEAKING         = "audio_channel_user_speaking"
	AUDIO_CHANNEL_USER_STOPPED_SPEAKING = "audio_channel_user_stopped_speaking"
)

type ZoneRTCPeerConnection struct {
	*webrtc.PeerConnection
	makingOffer     bool
	makingOfferLock *sync.Mutex
	negotiate       func(string, SendDCMessageFunc)
}

type AudioChannel struct {
	ID                       string                                `json:"id"`
	Owner                    string                                `json:"owner"`
	ChannelType              string                                `json:"channelType"`
	Members                  []string                              `json:"members"`
	CurrentMembersId         []string                              `json:"currentMembersId"`
	CurrentMembers           map[string]*AudioChannelMember        `json:"currentMembers"`
	localSD                  map[string]*webrtc.SessionDescription `json:"-"`
	rtcPeerConnections       map[string]*ZoneRTCPeerConnection     `json:"-"`
	audioTransceiver         map[string][]*PeerSender              `json:"-"`
	audiChannelsDataChannels map[string]*DataChannel               `json:"-"`
	pendingCandidates        map[string][]*webrtc.ICECandidate     `json:"-"`
	remoteTracks             map[string][]*RemoteTrack             `json:"-"`
	middlewares              []interface{}                         `json:"-"`
	candidateFlag            *uint32                               `json:"-"`
	remoteTracksFlag         *uint32                               `json:"-"`
	rtcPeerConnectionMapFlag *uint32                               `json:"-"`
	dataChannelMapFlag       *uint32                               `json:"-"`
	localSDMapFlag           *uint32                               `json:"-"`
	audioSenderFlag          *uint32                               `json:"-"`
}

type AudioChannelOnICECandidateFunc = func(string, string, *webrtc.ICECandidate) error

func NewAudioChannel(id string, owner string, channelType string, members []string, currentMembersId []string, currentMembers map[string]*AudioChannelMember) (audioChannel *AudioChannel) {
	candidateFlag := uint32(0)
	remoteTracksFlag := uint32(0)
	rtcPeerConnectionMapFlag := uint32(0)
	dataChannelMapFlag := uint32(0)
	localSDMapFlag := uint32(0)
	audioSenderFlag := uint32(0)
	audioChannel = &AudioChannel{
		ID:                       id,
		Owner:                    owner,
		ChannelType:              channelType,
		Members:                  members,
		CurrentMembersId:         currentMembersId,
		CurrentMembers:           currentMembers,
		localSD:                  make(map[string]*webrtc.SessionDescription),
		rtcPeerConnections:       make(map[string]*ZoneRTCPeerConnection),
		audioTransceiver:         make(map[string][]*PeerSender),
		audiChannelsDataChannels: make(map[string]*DataChannel),
		pendingCandidates:        make(map[string][]*webrtc.ICECandidate),
		remoteTracks:             make(map[string][]*RemoteTrack),
		middlewares:              make([]interface{}, 0),
		candidateFlag:            &candidateFlag,
		remoteTracksFlag:         &remoteTracksFlag,
		rtcPeerConnectionMapFlag: &rtcPeerConnectionMapFlag,
		dataChannelMapFlag:       &dataChannelMapFlag,
		localSDMapFlag:           &localSDMapFlag,
		audioSenderFlag:          &audioSenderFlag,
	}
	return
}

func (ac *AudioChannel) HandleOffer(ctx context.Context, channelId string, userId string, sdp string, hostId string, sendDCMessage SendDCMessageFunc, cb AudioChannelOnICECandidateFunc) (done chan struct{}, errCh chan error) {
	done, errCh = make(chan struct{}), make(chan error)
	go func() {
		peerConnection, err := ac.createPeerConnection(userId, ac.ID, webrtc.SDPTypeAnswer, cb, sendDCMessage)
		if err != nil {
			errCh <- err
			return
		}
		_ = atomicallyExecute(ac.rtcPeerConnectionMapFlag, func() (err error) {
			ac.rtcPeerConnections[userId] = &ZoneRTCPeerConnection{
				PeerConnection:  peerConnection,
				makingOffer:     false,
				makingOfferLock: &sync.Mutex{},
				negotiate:       ac.negotiate,
			}
			return
		})
		offer := webrtc.SessionDescription{
			Type: webrtc.SDPTypeOffer,
			SDP:  sdp,
		}
		if err = peerConnection.SetRemoteDescription(offer); err != nil {
			errCh <- err
			return
		}
		rawAnswer, err := peerConnection.CreateAnswer(nil)
		if err != nil {
			errCh <- err
			return
		}
		if err = peerConnection.SetLocalDescription(rawAnswer); err != nil {
			errCh <- err
			return
		}
		_, _ = sendDCMessage(string(AUDIO_CHANNEL_WEBRTC_ANSWER), hostId, userId, map[string]interface{}{
			"to":        userId,
			"from":      ac.ID,
			"channelId": channelId,
			"sdp":       rawAnswer.SDP,
		})
		done <- struct{}{}
		// logger.Println("handle offer done")
	}()
	return
}

func (ac *AudioChannel) HandleCounterOffer(ctx context.Context, userId string, sendDCMessage SendDCMessageFunc) (err error) {
	// if err = atomicallyExecute(ac.rtcPeerConnectionMapFlag, func() (err error) {
	// 	if _, ok := ac.rtcPeerConnections[userId]; !ok {
	// 		err = fmt.Errorf("no field corresponding peer connection for id %s", userId)
	// 		return
	// 	}
	// 	// logger.Println("handling counter offer")
	// connection := ac.rtcPeerConnections[userId]
	// err = atomicallyExecute(ac.localSDMapFlag, func() (err error) {
	// 	err = connection.SetLocalDescription(*ac.localSD[userId])
	// 	return
	// })
	// 	return
	// }); err != nil {
	// 	return
	// }
	// _ = atomicallyExecute(ac.localSDMapFlag, func() (err error) {
	// 	delete(ac.localSD, userId)
	// 	return
	// })
	if err = atomicallyExecute(ac.candidateFlag, func() (err error) {
		for _, candidate := range ac.pendingCandidates[userId] {
			// logger.Println("sending candidate to", userId, candidate)
			d, e := sendDCMessage(string(AUDIO_CHANNEL_WEBRTC_CANDIDATE), "", userId, map[string]interface{}{
				"from":          ac.ID,
				"to":            userId,
				"candidate":     candidate.ToJSON().Candidate,
				"sdpMid":        *candidate.ToJSON().SDPMid,
				"sdpMLineIndex": strconv.Itoa(int(*candidate.ToJSON().SDPMLineIndex)),
			})
			select {
			case <-d:
			case err = <-e:
				return
			}
		}
		delete(ac.pendingCandidates, userId)
		return
	}); err != nil {
		return
	}
	return
}

func (ac *AudioChannel) HandleRennegotiationOffer(from string, sdp string, sendDCMessage SendDCMessageFunc) (err error) {
	err = atomicallyExecute(ac.rtcPeerConnectionMapFlag, func() (err error) {
		if _, ok := ac.rtcPeerConnections[from]; !ok {
			err = fmt.Errorf("no corresponding peer connection for id %s", from)
			return
		}
		ac.rtcPeerConnections[from].makingOfferLock.Lock()
		if ac.rtcPeerConnections[from].makingOffer {
			ac.rtcPeerConnections[from].makingOfferLock.Unlock()
			return fmt.Errorf("already making an offer or state is stable")
		}
		ac.rtcPeerConnections[from].makingOfferLock.Unlock()
		if err = ac.rtcPeerConnections[from].SetRemoteDescription(webrtc.SessionDescription{SDP: sdp, Type: webrtc.SDPTypeOffer}); err != nil {
			return
		}
		localSd, err := ac.rtcPeerConnections[from].CreateAnswer(nil)
		if err != nil {
			return
		}
		if err = ac.rtcPeerConnections[from].SetLocalDescription(localSd); err != nil {
			return
		}
		d, e := sendDCMessage(string(AUDIO_CHANNEL_WEBRTC_RENNEGOTIATION_ANSWER), ac.ID, from, map[string]interface{}{
			"from": ac.ID,
			"to":   from,
			"sdp":  localSd.SDP,
		})
		select {
		case <-d:
		case err = <-e:
		}
		return
	})
	return
}

func (ac *AudioChannel) HandleRennegotiationAnswer(from string, sdp string) (err error) {
	// logger.Println("---------------------handling rennego answer")
	err = atomicallyExecute(ac.rtcPeerConnectionMapFlag, func() (err error) {
		ac.rtcPeerConnections[from].makingOfferLock.Lock()
		if ac.rtcPeerConnections[from].makingOffer {
			ac.rtcPeerConnections[from].makingOfferLock.Unlock()
			return fmt.Errorf("already making an offer or state is stable")
		}
		ac.rtcPeerConnections[from].makingOfferLock.Unlock()
		if _, ok := ac.rtcPeerConnections[from]; !ok {
			err = fmt.Errorf("no corresponding peer connection for id %s", from)
			return
		}
		err = ac.rtcPeerConnections[from].SetRemoteDescription(webrtc.SessionDescription{SDP: sdp, Type: webrtc.SDPTypeAnswer})
		return
	})
	return
}

func (ac *AudioChannel) AddCandidate(candidate *webrtc.ICECandidateInit, from string) (err error) {
	// logger.Println("adding ice candidate", candidate)
	err = atomicallyExecute(ac.rtcPeerConnectionMapFlag, func() (err error) {
		if _, ok := ac.rtcPeerConnections[from]; ok && candidate != nil {
			err = ac.rtcPeerConnections[from].AddICECandidate(*candidate)
		}
		return
	})
	return
}

func (ac *AudioChannel) createPeerConnection(target string, from string, peerType webrtc.SDPType, cb AudioChannelOnICECandidateFunc, sendDCMessage SendDCMessageFunc) (peerConnection *webrtc.PeerConnection, err error) {
	defer func() {
		if r := recover(); err != nil {
			logger.Printf("recover from panic : %v\n", r)
		}
	}()
	config := webrtc.Configuration{
		ICEServers: []webrtc.ICEServer{
			{
				URLs: []string{"stun:stun.l.google.com:19302"},
			},
		},
		SDPSemantics: webrtc.SDPSemanticsUnifiedPlanWithFallback,
	}

	peerConnection, err = webrtc.NewPeerConnection(config)
	if err != nil {
		return
	}
	// logger.Println("---------------------------------------------------")
	if peerType == webrtc.SDPTypeAnswer {
		maxRetransmits := uint16(100)
		channel, err := peerConnection.CreateDataChannel("data", &webrtc.DataChannelInit{
			MaxRetransmits: &maxRetransmits,
		})
		if err != nil {
			return nil, err
		}
		channel.OnOpen(func() {
			// logger.Println("channel opened")
			if chanErr := channel.SendText("yooo man this is open"); chanErr != nil {
				// logger.Println(chanErr)
			}
		})
		channel.OnMessage(func(msg webrtc.DataChannelMessage) {
			var event CallEvent
			if err := json.Unmarshal(msg.Data, &event); err != nil {
				// logger.Println(err)
				return
			}
			if e := ac.HandleDataChannelEvents(event.From, event.EventId, event.Payload); e != nil {
				// logger.Println("*-------------- datachannel error: ", e)
			}
		})
		// logger.Println("new channel for target : ", target)
		channel.SetBufferedAmountLowThreshold(bufferedAmountLowThreshold)
		channel.OnBufferedAmountLow(func() {

		})
		_ = atomicallyExecute(ac.dataChannelMapFlag, func() (err error) {
			// logger.Println(target)
			l := uint32(0)
			ac.audiChannelsDataChannels[target] = &DataChannel{
				DataChannel: channel,
				l:           &l,
			}
			return
		})
	} else {
		peerConnection.OnDataChannel(func(dc *webrtc.DataChannel) {
			_ = atomicallyExecute(ac.dataChannelMapFlag, func() (err error) {
				l := uint32(0)
				ac.audiChannelsDataChannels[target] = &DataChannel{
					DataChannel: dc,
					l:           &l,
				}
				return
			})
			dc.OnOpen(func() {
				// logger.Printf("got a new open datachannel %s\n", dc.Label())
			})
			dc.OnMessage(func(msg webrtc.DataChannelMessage) {
				var event CallEvent
				if err := json.Unmarshal(msg.Data, &event); err != nil {
					// logger.Println(err)
					return
				}
				if e := ac.HandleDataChannelEvents(event.From, event.EventId, event.Payload); e != nil {
					// logger.Println("*-------------- datachannel error: ", e)
				}
			})
		})
	}

	peerConnection.OnConnectionStateChange(func(pcs webrtc.PeerConnectionState) {
		if pcs == webrtc.PeerConnectionStateClosed || pcs == webrtc.PeerConnectionStateDisconnected || pcs == webrtc.PeerConnectionStateFailed {
			// logger.Println(pcs)
			//ac.HandleLeavingMember(target, squadId)
		}
	})
	peerConnection.OnICEConnectionStateChange(func(is webrtc.ICEConnectionState) {
		// logger.Printf("ICE connection state has changed %s\n", is.String())
		if is == webrtc.ICEConnectionStateDisconnected || is == webrtc.ICEConnectionStateFailed {
			// logger.Println(is)
		}
	})
	peerConnection.OnTrack(func(tr *webrtc.TrackRemote, r *webrtc.RTPReceiver) {
		err = atomicallyExecute(ac.rtcPeerConnectionMapFlag, func() (err error) {
			err = atomicallyExecute(ac.remoteTracksFlag, func() (err error) {
				// logger.Println("------------------", ac.CurrentMembersId)
				for _, id := range ac.CurrentMembersId {
					// logger.Println(id)
					if id != target {
						if _, ok := ac.remoteTracks[id]; !ok {
							continue
						}
						for _, track := range ac.remoteTracks[id] {
							transceiver, err := peerConnection.AddTransceiverFromTrack(track.Track)
							if err != nil {
								// logger.Println("add track error")
								continue
							}
							if err := transceiver.Sender().ReplaceTrack(track.Track); err != nil {
								// logger.Println("add track error")
								continue
							}

							_ = atomicallyExecute(ac.audioSenderFlag, func() (err error) {
								if len(ac.audioTransceiver) == 0 {
									ac.audioTransceiver[id] = []*PeerSender{{ID: target, Transceiver: transceiver}}
								} else {
									ac.audioTransceiver[id] = append(ac.audioTransceiver[id], &PeerSender{ID: target, Transceiver: transceiver})
								}
								return
							})
							// logger.Println("track added", track)
						}
					}
				}
				return
			})
			return
		})
		// logger.Println("got new track")
		defer func() {
			if stopErr := r.Stop(); stopErr != nil {
				// logger.Println(stopErr)
			}
		}()
		go func() {
			ticker := time.NewTicker(1500 * time.Millisecond)
			for range ticker.C {
				if rtcpSendErr := peerConnection.WriteRTCP([]rtcp.Packet{&rtcp.PictureLossIndication{MediaSSRC: uint32(tr.SSRC())}}); rtcpSendErr != nil {
					// logger.Println(rtcpSendErr)
					break
				}
			}
		}()
		uniqId := uuid.New()
		i := fmt.Sprintf("%s/%s", target, uniqId.String())
		// logger.Println("*************************----------------", i, "-----------------------***************")
		localTrack, newTrackErr := webrtc.NewTrackLocalStaticRTP(tr.Codec().RTPCodecCapability, i, i)
		if newTrackErr != nil {
			return
		}
		// logger.Println(localTrack)
		rtpbuf := make([]byte, 1400)
		flag := int32(0)
		remote := &RemoteTrack{ID: target, Track: localTrack, rdv: &flag}
		_ = atomicallyExecute(ac.remoteTracksFlag, func() (err error) {
			if len(ac.remoteTracks[target]) == 0 {
				ac.remoteTracks[target] = []*RemoteTrack{remote}
			} else {
				ac.remoteTracks[target] = append(ac.remoteTracks[target], remote)
			}
			_ = len(ac.remoteTracks[target])
			// logger.Println(index, ac.remoteTracks)
			return
		})
		_ = atomicallyExecute(ac.rtcPeerConnectionMapFlag, func() (err error) {
			for _, id := range ac.CurrentMembersId {
				if id != target {
					if _, ok := ac.rtcPeerConnections[id]; !ok {
						continue
					}
					connection := ac.rtcPeerConnections[id]
					transceiver, tranceiverErr := connection.AddTransceiverFromTrack(localTrack)
					if tranceiverErr != nil {
						// logger.Println(tranceiverErr)
						continue
					}
					if replaceTrackErr := transceiver.Sender().ReplaceTrack(localTrack); replaceTrackErr != nil {
						// logger.Println(replaceTrackErr)
						continue
					}
					go func() {
						rtcpBuf := make([]byte, 1500)
						for {
							if _, _, rtcpErr := transceiver.Sender().Read(rtcpBuf); rtcpErr != nil {
								return
							}
						}
					}()
					if localTrack.Kind() == webrtc.RTPCodecTypeAudio {
						_ = atomicallyExecute(ac.audioSenderFlag, func() (err error) {
							if len(ac.audioTransceiver) == 0 {
								ac.audioTransceiver[target] = []*PeerSender{{ID: id, Transceiver: transceiver}}
							} else {
								ac.audioTransceiver[target] = append(ac.audioTransceiver[target], &PeerSender{ID: id, Transceiver: transceiver})
							}
							return
						})
					} else if localTrack.Kind() == webrtc.RTPCodecTypeVideo {
						// logger.Println("track of wrong type")
					}
					go func(id string) {

						connection.negotiate(id, sendDCMessage)
					}(id)
				}
			}
			return
		})
		d := make(chan struct{})
		go func() {
			for {
				i, _, readErr := tr.Read(rtpbuf)
				if readErr != nil {
					// logger.Println(readErr)
					break
				}
				//// logger.Println(rtpbuf[:i])
				f := atomic.LoadInt32(remote.rdv)
				if f == 0 {
					if _, writeErr := localTrack.Write(rtpbuf[:i]); writeErr != nil && !errors.Is(writeErr, io.ErrClosedPipe) {
						// logger.Println(writeErr)
						break
					} else {
						_ = rtpbuf[:i]
					}
				}
			}
			d <- struct{}{}
		}()
		<-d
	})
	peerConnection.OnICECandidate(func(i *webrtc.ICECandidate) {
		if i == nil {
			return
		}
		_ = atomicallyExecute(ac.candidateFlag, func() (err error) {
			desc := peerConnection.RemoteDescription()
			if desc == nil {
				// logger.Println("generated candidate appended to list : ", i)
				ac.pendingCandidates[target] = append(ac.pendingCandidates[target], i)
			} else {
				// logger.Println("generated candidate : ", i)
				if iceCandidateErr := cb(from, target, i); iceCandidateErr != nil {
					// logger.Println(iceCandidateErr)
				}
			}
			return
		})
	})
	//peerConnection.OnNegotiationNeeded(func() {
	//// logger.Println("---------------- rennego is needed -----------")
	// _ = atomicallyExecute(ac.rtcPeerConnectionMapFlag, func() (err error) {
	// 	for _, id := range ac.CurrentMembersId {
	// 		// logger.Println("----------------- sending renego to peer with id", id)
	// 		if _, ok := ac.rtcPeerConnections[id]; !ok {
	// 			continue
	// 		}
	// 		if peerConnection.SignalingState() == webrtc.SignalingStateStable {
	// 			localSd, localSdErr := peerConnection.CreateOffer(nil)
	// 			if localSdErr != nil {
	// 				// logger.Println(localSdErr)
	// 				return localSdErr
	// 			}
	// 			if err = peerConnection.SetLocalDescription(localSd); err != nil {
	// 				// logger.Println(err)
	// 				return
	// 			}
	// 			d, e := sendDCMessage(string(AUDIO_CHANNEL_WEBRTC_RENNEGOTIATION_OFFER), ac.ID, id, map[string]interface{}{
	// 				"from": ac.ID,
	// 				"to":   id,
	// 				"sdp":  localSd.SDP,
	// 			})
	// 			select {
	// 			case <-d:
	// 			case err = <-e:
	// 				// logger.Println(err)
	// 			}
	// 		}
	// 	}
	// 	return
	// })
	//})
	return
}

func (ac *AudioChannel) HandleLeavingMember(id string) {
	if err := atomicallyExecute(ac.rtcPeerConnectionMapFlag, func() (err error) {
		if _, ok := ac.rtcPeerConnections[id]; !ok {
			err = fmt.Errorf("no corresponding peerconnection for audio channel leaving member")
		}
		return
	}); err != nil {
		// logger.Println(err)
	} else {
		defer func() {
			_ = atomicallyExecute(ac.rtcPeerConnectionMapFlag, func() (err error) {
				if _, ok := ac.rtcPeerConnections[id]; ok {
					if closeErr := ac.rtcPeerConnections[id].Close(); closeErr != nil {
						err = closeErr
						// logger.Println("peer connection close error", closeErr)
					}
				}
				delete(ac.rtcPeerConnections, id)
				return
			})
		}()
	}
	// logger.Printf("peer %s is leaving the squad\n", id)
	_ = atomicallyExecute(ac.dataChannelMapFlag, func() (err error) {
		if _, ok := ac.audiChannelsDataChannels[id]; ok {
			ac.audiChannelsDataChannels[id].DataChannel.Close()
		}
		delete(ac.audiChannelsDataChannels, id)
		return
	})
	_ = atomicallyExecute(ac.localSDMapFlag, func() (err error) {
		delete(ac.localSD, id)
		return
	})
	_ = atomicallyExecute(ac.candidateFlag, func() (err error) {
		delete(ac.pendingCandidates, id)
		return
	})
	_ = atomicallyExecute(ac.audioSenderFlag, func() (err error) {
		for peerId, peerSender := range ac.audioTransceiver {
			if peerId != id {
				// logger.Println("senders", peerSender)
				c := 0
				for i, sender := range peerSender {
					if sender.ID == id {
						if senderErr := sender.Transceiver.Sender().Stop(); senderErr != nil {
							// logger.Println(senderErr)
						}
						if transceiverErr := sender.Transceiver.Stop(); transceiverErr != nil {
							// logger.Println("transceiverErr occured with video", transceiverErr)
						}
						peerSender[len(peerSender)-i-1], peerSender[i] = peerSender[i], peerSender[len(peerSender)-i-1]
						c++
					}
				}
				ac.audioTransceiver[peerId] = ac.audioTransceiver[peerId][:len(peerSender)-(c)]
				// logger.Println(ac.audioTransceiver[peerId])
			}
		}
		for _, transceiver := range ac.audioTransceiver[id] {
			if senderErr := transceiver.Transceiver.Sender().Stop(); senderErr != nil {
				// logger.Println(senderErr)
			}
			if stopErr := transceiver.Transceiver.Stop(); stopErr != nil {
				// logger.Println("transceiver audio stop error", stopErr)
			}
		}
		delete(ac.audioTransceiver, id)
		return
	})
	_ = atomicallyExecute(ac.remoteTracksFlag, func() (err error) {
		delete(ac.remoteTracks, id)
		return
	})
}

func (ac *AudioChannel) negotiate(target string, sendDCMessage SendDCMessageFunc) {
	// logger.Println("------------------negotiate is called")
	_ = atomicallyExecute(ac.rtcPeerConnectionMapFlag, func() (err error) {
		if _, ok := ac.rtcPeerConnections[target]; !ok {
			return
		}
		ac.rtcPeerConnections[target].makingOfferLock.Lock()
		ac.rtcPeerConnections[target].makingOffer = true
		ac.rtcPeerConnections[target].makingOfferLock.Unlock()
		defer func() {
			ac.rtcPeerConnections[target].makingOfferLock.Lock()
			ac.rtcPeerConnections[target].makingOffer = false
			ac.rtcPeerConnections[target].makingOfferLock.Unlock()
		}()

		for _, id := range ac.CurrentMembersId {
			// logger.Println("----------------- sending renego to peer with id", id)
			if _, ok := ac.rtcPeerConnections[id]; !ok {
				continue
			}
			connection := ac.rtcPeerConnections[id]
			if connection.SignalingState() == webrtc.SignalingStateStable {
				localSd, err := connection.CreateOffer(nil)
				if err != nil {
					// logger.Println(err)
					return err
				}
				if err = connection.SetLocalDescription(localSd); err != nil {
					// logger.Println(err)
					return err
				}
				d, e := sendDCMessage(string(AUDIO_CHANNEL_WEBRTC_RENNEGOTIATION_OFFER), ac.ID, id, map[string]interface{}{
					"from": ac.ID,
					"to":   id,
					"sdp":  localSd.SDP,
				})
				select {
				case <-d:
				case err = <-e:
					// logger.Println(err)
				}
			}
		}
		return
	})
}

func (ac *AudioChannel) broadcastDatachannelMessage(from string, eventId string, payload map[string]interface{}) (done chan struct{}, errCh chan error) {
	done, errCh = make(chan struct{}), make(chan error)
	go func() {
		bs, jsonErr := json.Marshal(&ZoneResponse{
			Type:    eventId,
			From:    ac.ID,
			Payload: payload,
		})
		if jsonErr != nil {
			errCh <- jsonErr
			return
		}
		if err := atomicallyExecute(ac.dataChannelMapFlag, func() (err error) {
			for id, dc := range ac.audiChannelsDataChannels {
				if from != id {
					if err = dc.DataChannel.SendText(string(bs)); err != nil {
						return
					}
				}
			}
			return
		}); err != nil {
			errCh <- err
		}
		done <- struct{}{}
	}()
	return
}

func (ac *AudioChannel) HandleDataChannelEvents(from string, eventId string, payload map[string]interface{}) (err error) {
	switch eventId {
	case AUDIO_CHANNEL_USER_MUTE:
		if err = atomicallyExecute(ac.remoteTracksFlag, func() (err error) {
			if _, ok := ac.remoteTracks[from]; !ok {
				err = fmt.Errorf("no corresponding remote tracks entry for id %s", from)
				return
			}
			for _, track := range ac.remoteTracks[from] {
				atomic.SwapInt32(track.rdv, 1)
			}
			return
		}); err != nil {
			return
		}
		done, errCh := ac.broadcastDatachannelMessage(from, AUDIO_CHANNEL_USER_MUTE, map[string]interface{}{
			"userId": from,
		})
		select {
		case <-done:
		case err = <-errCh:
		}
	case AUDIO_CHANNEL_USER_UNMUTE:
		if err = atomicallyExecute(ac.remoteTracksFlag, func() (err error) {
			if _, ok := ac.remoteTracks[from]; !ok {
				err = fmt.Errorf("no corresponding remote tracks entry for id %s", from)
				return
			}
			for _, track := range ac.remoteTracks[from] {
				atomic.SwapInt32(track.rdv, 0)
			}
			return
		}); err != nil {
			return
		}
		done, errCh := ac.broadcastDatachannelMessage(from, AUDIO_CHANNEL_USER_UNMUTE, map[string]interface{}{
			"userId": from,
		})
		select {
		case <-done:
		case err = <-errCh:
		}
	case AUDIO_CHANNEL_USER_SPEAKING:
		done, errCh := ac.broadcastDatachannelMessage(from, AUDIO_CHANNEL_USER_SPEAKING, map[string]interface{}{
			"userId": from,
		})
		select {
		case <-done:
		case err = <-errCh:
		}
	case AUDIO_CHANNEL_USER_STOPPED_SPEAKING:
		done, errCh := ac.broadcastDatachannelMessage(from, AUDIO_CHANNEL_USER_STOPPED_SPEAKING, map[string]interface{}{
			"userId": from,
		})
		select {
		case <-done:
		case err = <-errCh:
		}
	}
	return
}