Add SIP Support (#2240)

pkg/config/config.go

@@ -66,6 +66,7 @@ type Config struct {
 	Room           RoomConfig               `yaml:"room,omitempty"`
 	TURN           TURNConfig               `yaml:"turn,omitempty"`
 	Ingress        IngressConfig            `yaml:"ingress,omitempty"`
+	SIP            SIPConfig                `yaml:"sip,omitempty"`
 	WebHook        WebHookConfig            `yaml:"webhook,omitempty"`
 	NodeSelector   NodeSelectorConfig       `yaml:"node_selector,omitempty"`
 	KeyFile        string                   `yaml:"key_file,omitempty"`
@@ -294,6 +295,9 @@ type IngressConfig struct {
 	WHIPBaseURL string `yaml:"whip_base_url,omitempty"`
 }
 
+type SIPConfig struct {
+}
+
 // not exposed to YAML
 type APIConfig struct {
 	// amount of time to wait for API to execute, default 2s

pkg/service/errors.go

@@ -34,4 +34,8 @@ var (
 	ErrRemoteUnmuteNoteEnabled = psrpc.NewErrorf(psrpc.FailedPrecondition, "remote unmute not enabled")
 	ErrTrackNotFound           = psrpc.NewErrorf(psrpc.NotFound, "track is not found")
 	ErrWebHookMissingAPIKey    = psrpc.NewErrorf(psrpc.InvalidArgument, "api_key is required to use webhooks")
+	ErrSIPNotConnected         = psrpc.NewErrorf(psrpc.Internal, "sip not connected (redis required)")
+	ErrSIPTrunkNotFound        = psrpc.NewErrorf(psrpc.NotFound, "requested sip trunk does not exist")
+	ErrSIPDispatchRuleNotFound = psrpc.NewErrorf(psrpc.NotFound, "requested sip dispatch rule does not exist")
+	ErrSIPParticipantNotFound  = psrpc.NewErrorf(psrpc.NotFound, "requested sip participant does not exist")
 )

pkg/service/interfaces.go

@@ -76,3 +76,21 @@ type RoomAllocator interface {
 	CreateRoom(ctx context.Context, req *livekit.CreateRoomRequest) (*livekit.Room, bool, error)
 	ValidateCreateRoom(ctx context.Context, roomName livekit.RoomName) error
 }
+
+//counterfeiter:generate . SIPStore
+type SIPStore interface {
+	StoreSIPTrunk(ctx context.Context, info *livekit.SIPTrunkInfo) error
+	LoadSIPTrunk(ctx context.Context, sipTrunkID string) (*livekit.SIPTrunkInfo, error)
+	ListSIPTrunk(ctx context.Context) ([]*livekit.SIPTrunkInfo, error)
+	DeleteSIPTrunk(ctx context.Context, info *livekit.SIPTrunkInfo) error
+
+	StoreSIPDispatchRule(ctx context.Context, info *livekit.SIPDispatchRuleInfo) error
+	LoadSIPDispatchRule(ctx context.Context, sipDispatchRuleID string) (*livekit.SIPDispatchRuleInfo, error)
+	ListSIPDispatchRule(ctx context.Context) ([]*livekit.SIPDispatchRuleInfo, error)
+	DeleteSIPDispatchRule(ctx context.Context, info *livekit.SIPDispatchRuleInfo) error
+
+	StoreSIPParticipant(ctx context.Context, info *livekit.SIPParticipantInfo) error
+	LoadSIPParticipant(ctx context.Context, sipParticipantID string) (*livekit.SIPParticipantInfo, error)
+	ListSIPParticipant(ctx context.Context) ([]*livekit.SIPParticipantInfo, error)
+	DeleteSIPParticipant(ctx context.Context, info *livekit.SIPParticipantInfo) error
+}

pkg/service/ioservice.go

@@ -17,6 +17,7 @@ package service
 import (
 	"context"
 	"errors"
+	"fmt"
 
 	"google.golang.org/protobuf/types/known/emptypb"
 
@@ -32,6 +33,7 @@ type IOInfoService struct {
 
 	es        EgressStore
 	is        IngressStore
+	ss        SIPStore
 	telemetry telemetry.TelemetryService
 
 	shutdown chan struct{}
@@ -41,11 +43,13 @@ func NewIOInfoService(
 	bus psrpc.MessageBus,
 	es EgressStore,
 	is IngressStore,
+	ss SIPStore,
 	ts telemetry.TelemetryService,
 ) (*IOInfoService, error) {
 	s := &IOInfoService{
 		es:        es,
 		is:        is,
+		ss:        ss,
 		telemetry: ts,
 		shutdown:  make(chan struct{}),
 	}
@@ -211,6 +215,31 @@ func (s *IOInfoService) UpdateIngressState(ctx context.Context, req *rpc.UpdateI
 	return &emptypb.Empty{}, nil
 }
 
+func (s *IOInfoService) EvaluateSIPDispatchRules(ctx context.Context, req *rpc.EvaluateSIPDispatchRulesRequest) (*rpc.EvaluateSIPDispatchRulesResponse, error) {
+	dispatchRules, err := s.ss.ListSIPDispatchRule(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(dispatchRules) == 0 {
+		return nil, fmt.Errorf("No SIP Dispatch Rule Found")
+	}
+
+	directDispatchRule := dispatchRules[0].Rule.GetDispatchRuleDirect()
+	if directDispatchRule == nil {
+		return nil, fmt.Errorf("No SIP Direct Dispatch Rule Found")
+	}
+
+	return &rpc.EvaluateSIPDispatchRulesResponse{
+		RoomName:            directDispatchRule.RoomName,
+		ParticipantIdentity: req.CallingNumber,
+	}, nil
+}
+
+func (s *IOInfoService) GetSIPTrunkAuthentication(ctx context.Context, req *rpc.GetSIPTrunkAuthenticationRequest) (*rpc.GetSIPTrunkAuthenticationResponse, error) {
+	return nil, nil
+}
+
 func (s *IOInfoService) Stop() {
 	close(s.shutdown)
 

pkg/service/redisstore.go

@@ -51,6 +51,10 @@ const (
 	IngressStatePrefix = "{ingress}_state:"
 	RoomIngressPrefix  = "room_{ingress}:"
 
+	SIPTrunkKey        = "sip_trunk"
+	SIPDispatchRuleKey = "sip_dispatch_rule"
+	SIPParticipantKey  = "sip_participant"
+
 	// RoomParticipantsPrefix is hash of participant_name => ParticipantInfo
 	RoomParticipantsPrefix = "room_participants:"
 
@@ -812,3 +816,129 @@ func (s *RedisStore) DeleteIngress(_ context.Context, info *livekit.IngressInfo)
 
 	return nil
 }
+
+func (s *RedisStore) loadOne(ctx context.Context, key, id string, info proto.Message, notFoundErr error) error {
+	data, err := s.rc.HGet(s.ctx, key, id).Result()
+	switch err {
+	case nil:
+		return proto.Unmarshal([]byte(data), info)
+	case redis.Nil:
+		return notFoundErr
+	default:
+		return err
+	}
+}
+
+func (s *RedisStore) loadMany(ctx context.Context, key string, onResult func() proto.Message) error {
+	data, err := s.rc.HGetAll(s.ctx, key).Result()
+	if err != nil {
+		if err == redis.Nil {
+			return nil
+		}
+		return err
+	}
+
+	for _, d := range data {
+		if err = proto.Unmarshal([]byte(d), onResult()); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (s *RedisStore) StoreSIPTrunk(ctx context.Context, info *livekit.SIPTrunkInfo) error {
+	data, err := proto.Marshal(info)
+	if err != nil {
+		return err
+	}
+
+	return s.rc.HSet(s.ctx, SIPTrunkKey, info.SipTrunkId, data).Err()
+}
+
+func (s *RedisStore) LoadSIPTrunk(ctx context.Context, sipTrunkId string) (*livekit.SIPTrunkInfo, error) {
+	info := &livekit.SIPTrunkInfo{}
+	if err := s.loadOne(ctx, SIPTrunkKey, sipTrunkId, info, ErrSIPTrunkNotFound); err != nil {
+		return nil, err
+	}
+
+	return info, nil
+}
+
+func (s *RedisStore) DeleteSIPTrunk(ctx context.Context, info *livekit.SIPTrunkInfo) error {
+	return s.rc.HDel(s.ctx, SIPTrunkKey, info.SipTrunkId).Err()
+}
+
+func (s *RedisStore) ListSIPTrunk(ctx context.Context) (infos []*livekit.SIPTrunkInfo, err error) {
+	err = s.loadMany(ctx, SIPTrunkKey, func() proto.Message {
+		infos = append(infos, &livekit.SIPTrunkInfo{})
+		return infos[len(infos)-1]
+	})
+
+	return infos, err
+}
+
+func (s *RedisStore) StoreSIPDispatchRule(ctx context.Context, info *livekit.SIPDispatchRuleInfo) error {
+	data, err := proto.Marshal(info)
+	if err != nil {
+		return err
+	}
+
+	return s.rc.HSet(s.ctx, SIPDispatchRuleKey, info.SipDispatchRuleId, data).Err()
+}
+
+func (s *RedisStore) LoadSIPDispatchRule(ctx context.Context, sipDispatchRuleId string) (*livekit.SIPDispatchRuleInfo, error) {
+	info := &livekit.SIPDispatchRuleInfo{}
+	if err := s.loadOne(ctx, SIPDispatchRuleKey, sipDispatchRuleId, info, ErrSIPDispatchRuleNotFound); err != nil {
+		return nil, err
+	}
+
+	return info, nil
+}
+
+func (s *RedisStore) DeleteSIPDispatchRule(ctx context.Context, info *livekit.SIPDispatchRuleInfo) error {
+	return s.rc.HDel(s.ctx, SIPDispatchRuleKey, info.SipDispatchRuleId).Err()
+}
+
+func (s *RedisStore) ListSIPDispatchRule(ctx context.Context) (infos []*livekit.SIPDispatchRuleInfo, err error) {
+	err = s.loadMany(ctx, SIPDispatchRuleKey, func() proto.Message {
+		infos = append(infos, &livekit.SIPDispatchRuleInfo{})
+		return infos[len(infos)-1]
+	})
+
+	return infos, err
+}
+
+func (s *RedisStore) StoreSIPParticipant(ctx context.Context, info *livekit.SIPParticipantInfo) error {
+	data, err := proto.Marshal(info)
+	if err != nil {
+		return err
+	}
+
+	return s.rc.HSet(s.ctx, SIPParticipantKey, info.SipParticipantId, data).Err()
+}
+func (s *RedisStore) LoadSIPParticipant(ctx context.Context, sipParticipantId string) (*livekit.SIPParticipantInfo, error) {
+	info := &livekit.SIPParticipantInfo{}
+	if err := s.loadOne(ctx, SIPParticipantKey, sipParticipantId, info, ErrSIPParticipantNotFound); err != nil {
+		return nil, err
+	}
+
+	return info, nil
+}
+
+func (s *RedisStore) DeleteSIPParticipant(ctx context.Context, info *livekit.SIPParticipantInfo) error {
+	return s.rc.HDel(s.ctx, SIPParticipantKey, info.SipParticipantId).Err()
+}
+
+func (s *RedisStore) ListSIPParticipant(ctx context.Context) (infos []*livekit.SIPParticipantInfo, err error) {
+	err = s.loadMany(ctx, SIPParticipantKey, func() proto.Message {
+		infos = append(infos, &livekit.SIPParticipantInfo{})
+		return infos[len(infos)-1]
+	})
+
+	return infos, err
+}
+
+func (s *RedisStore) SendSIPParticipantDTMF(ctx context.Context, info *livekit.SendSIPParticipantDTMFRequest) (*livekit.SIPParticipantDTMFInfo, error) {
+	return nil, fmt.Errorf("TODO")
+}

pkg/service/server.go

@@ -65,6 +65,7 @@ func NewLivekitServer(conf *config.Config,
 	roomService livekit.RoomService,
 	egressService *EgressService,
 	ingressService *IngressService,
+	sipService *SIPService,
 	ioService *IOInfoService,
 	rtcService *RTCService,
 	agentService *AgentService,
@@ -116,6 +117,7 @@ func NewLivekitServer(conf *config.Config,
 		),
 	))
 	ingressServer := livekit.NewIngressServer(ingressService, twirpLoggingHook)
+	sipServer := livekit.NewSIPServer(sipService, twirpLoggingHook)
 
 	mux := http.NewServeMux()
 	if conf.Development {
@@ -127,6 +129,7 @@ func NewLivekitServer(conf *config.Config,
 	mux.Handle(roomServer.PathPrefix(), roomServer)
 	mux.Handle(egressServer.PathPrefix(), egressServer)
 	mux.Handle(ingressServer.PathPrefix(), ingressServer)
+	mux.Handle(sipServer.PathPrefix(), sipServer)
 	mux.Handle("/rtc", rtcService)
 	mux.Handle("/agent", agentService)
 	mux.HandleFunc("/rtc/validate", rtcService.Validate)

pkg/service/sip.go

@@ -0,0 +1,207 @@
+// Copyright 2023 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package service
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/livekit/livekit-server/pkg/config"
+	"github.com/livekit/livekit-server/pkg/telemetry"
+	"github.com/livekit/protocol/livekit"
+	"github.com/livekit/protocol/rpc"
+	"github.com/livekit/protocol/utils"
+	"github.com/livekit/psrpc"
+)
+
+type SIPService struct {
+	conf        *config.SIPConfig
+	nodeID      livekit.NodeID
+	bus         psrpc.MessageBus
+	psrpcClient rpc.SIPClient
+	store       SIPStore
+	roomService livekit.RoomService
+}
+
+func NewSIPService(
+	conf *config.SIPConfig,
+	nodeID livekit.NodeID,
+	bus psrpc.MessageBus,
+	psrpcClient rpc.SIPClient,
+	store SIPStore,
+	rs livekit.RoomService,
+	ts telemetry.TelemetryService,
+) *SIPService {
+	return &SIPService{
+		conf:        conf,
+		nodeID:      nodeID,
+		bus:         bus,
+		psrpcClient: psrpcClient,
+		store:       store,
+		roomService: rs,
+	}
+}
+
+func (s *SIPService) CreateSIPTrunk(ctx context.Context, req *livekit.CreateSIPTrunkRequest) (*livekit.SIPTrunkInfo, error) {
+	if s.store == nil {
+		return nil, ErrSIPNotConnected
+	}
+
+	info := &livekit.SIPTrunkInfo{
+		SipTrunkId:          utils.NewGuid(utils.SIPTrunkPrefix),
+		InboundAddresses:    req.InboundAddresses,
+		OutboundAddress:     req.OutboundAddress,
+		OutboundNumber:      req.OutboundNumber,
+		InboundNumbersRegex: req.InboundNumbersRegex,
+		Username:            req.Username,
+		Password:            req.Password,
+	}
+
+	if err := s.store.StoreSIPTrunk(ctx, info); err != nil {
+		return nil, err
+	}
+	return info, nil
+}
+
+func (s *SIPService) ListSIPTrunk(ctx context.Context, req *livekit.ListSIPTrunkRequest) (*livekit.ListSIPTrunkResponse, error) {
+	if s.store == nil {
+		return nil, ErrSIPNotConnected
+	}
+
+	trunks, err := s.store.ListSIPTrunk(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	return &livekit.ListSIPTrunkResponse{Items: trunks}, nil
+}
+
+func (s *SIPService) DeleteSIPTrunk(ctx context.Context, req *livekit.DeleteSIPTrunkRequest) (*livekit.SIPTrunkInfo, error) {
+	if s.store == nil {
+		return nil, ErrSIPNotConnected
+	}
+
+	info, err := s.store.LoadSIPTrunk(ctx, req.SipTrunkId)
+	if err != nil {
+		return nil, err
+	}
+
+	if err = s.store.DeleteSIPTrunk(ctx, info); err != nil {
+		return nil, err
+	}
+
+	return info, nil
+}
+
+func (s *SIPService) CreateSIPDispatchRule(ctx context.Context, req *livekit.CreateSIPDispatchRuleRequest) (*livekit.SIPDispatchRuleInfo, error) {
+	if s.store == nil {
+		return nil, ErrSIPNotConnected
+	}
+
+	info := &livekit.SIPDispatchRuleInfo{
+		SipDispatchRuleId: utils.NewGuid(utils.SIPDispatchRulePrefix),
+		Rule:              req.Rule,
+		TrunkIds:          req.TrunkIds,
+		HidePhoneNumber:   req.HidePhoneNumber,
+	}
+
+	if err := s.store.StoreSIPDispatchRule(ctx, info); err != nil {
+		return nil, err
+	}
+	return info, nil
+}
+
+func (s *SIPService) ListSIPDispatchRule(ctx context.Context, req *livekit.ListSIPDispatchRuleRequest) (*livekit.ListSIPDispatchRuleResponse, error) {
+	if s.store == nil {
+		return nil, ErrSIPNotConnected
+	}
+
+	rules, err := s.store.ListSIPDispatchRule(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	return &livekit.ListSIPDispatchRuleResponse{Items: rules}, nil
+}
+
+func (s *SIPService) DeleteSIPDispatchRule(ctx context.Context, req *livekit.DeleteSIPDispatchRuleRequest) (*livekit.SIPDispatchRuleInfo, error) {
+	if s.store == nil {
+		return nil, ErrSIPNotConnected
+	}
+
+	info, err := s.store.LoadSIPDispatchRule(ctx, req.SipDispatchRuleId)
+	if err != nil {
+		return nil, err
+	}
+
+	if err = s.store.DeleteSIPDispatchRule(ctx, info); err != nil {
+		return nil, err
+	}
+
+	return info, nil
+}
+
+func (s *SIPService) CreateSIPParticipant(ctx context.Context, req *livekit.CreateSIPParticipantRequest) (*livekit.SIPParticipantInfo, error) {
+	if s.store == nil {
+		return nil, ErrSIPNotConnected
+	}
+
+	info := &livekit.SIPParticipantInfo{
+		SipParticipantId: utils.NewGuid(utils.SIPParticipantPrefix),
+	}
+
+	if err := s.store.StoreSIPParticipant(ctx, info); err != nil {
+		return nil, err
+	}
+	return info, nil
+}
+
+func (s *SIPService) ListSIPParticipant(ctx context.Context, req *livekit.ListSIPParticipantRequest) (*livekit.ListSIPParticipantResponse, error) {
+	if s.store == nil {
+		return nil, ErrSIPNotConnected
+	}
+
+	participants, err := s.store.ListSIPParticipant(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	return &livekit.ListSIPParticipantResponse{Items: participants}, nil
+}
+
+func (s *SIPService) DeleteSIPParticipant(ctx context.Context, req *livekit.DeleteSIPParticipantRequest) (*livekit.SIPParticipantInfo, error) {
+	if s.store == nil {
+		return nil, ErrSIPNotConnected
+	}
+
+	info, err := s.store.LoadSIPParticipant(ctx, req.SipParticipantId)
+	if err != nil {
+		return nil, err
+	}
+
+	if err = s.store.DeleteSIPParticipant(ctx, info); err != nil {
+		return nil, err
+	}
+
+	return info, nil
+}
+
+func (s *SIPService) SendSIPParticipantDTMF(ctx context.Context, req *livekit.SendSIPParticipantDTMFRequest) (*livekit.SIPParticipantDTMFInfo, error) {
+	if s.store == nil {
+		return nil, ErrSIPNotConnected
+	}
+
+	return nil, fmt.Errorf("TODO")
+}

pkg/service/wire.go

@@ -70,6 +70,10 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
 		getIngressStore,
 		getIngressConfig,
 		NewIngressService,
+		rpc.NewSIPClient,
+		getSIPStore,
+		getSIPConfig,
+		NewSIPService,
 		NewRoomAllocator,
 		NewRoomService,
 		NewRTCService,
@@ -195,6 +199,19 @@ func getIngressConfig(conf *config.Config) *config.IngressConfig {
 	return &conf.Ingress
 }
 
+func getSIPStore(s ObjectStore) SIPStore {
+	switch store := s.(type) {
+	case *RedisStore:
+		return store
+	default:
+		return nil
+	}
+}
+
+func getSIPConfig(conf *config.Config) *config.SIPConfig {
+	return &conf.SIP
+}
+
 func createClientConfiguration() clientconfiguration.ClientConfigurationManager {
 	return clientconfiguration.NewStaticClientConfigurationManager(clientconfiguration.StaticConfigurations)
 }

pkg/service/wire_gen.go

@@ -66,6 +66,7 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
 	}
 	egressStore := getEgressStore(objectStore)
 	ingressStore := getIngressStore(objectStore)
+	sipStore := getSIPStore(objectStore)
 	keyProvider, err := createKeyProvider(conf)
 	if err != nil {
 		return nil, err
@@ -76,7 +77,7 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
 	}
 	analyticsService := telemetry.NewAnalyticsService(conf, currentNode)
 	telemetryService := telemetry.NewTelemetryService(queuedNotifier, analyticsService)
-	ioInfoService, err := NewIOInfoService(messageBus, egressStore, ingressStore, telemetryService)
+	ioInfoService, err := NewIOInfoService(messageBus, egressStore, ingressStore, sipStore, telemetryService)
 	if err != nil {
 		return nil, err
 	}
@@ -102,6 +103,12 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
 		return nil, err
 	}
 	ingressService := NewIngressService(ingressConfig, nodeID, messageBus, ingressClient, ingressStore, roomService, telemetryService)
+	sipConfig := getSIPConfig(conf)
+	sipClient, err := rpc.NewSIPClient(messageBus)
+	if err != nil {
+		return nil, err
+	}
+	sipService := NewSIPService(sipConfig, nodeID, messageBus, sipClient, sipStore, roomService, telemetryService)
 	rtcService := NewRTCService(conf, roomAllocator, objectStore, router, currentNode, agentClient, telemetryService)
 	agentService, err := NewAgentService(messageBus)
 	if err != nil {
@@ -123,7 +130,7 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live
 	if err != nil {
 		return nil, err
 	}
-	livekitServer, err := NewLivekitServer(conf, roomService, egressService, ingressService, ioInfoService, rtcService, agentService, keyProvider, router, roomManager, signalServer, server, currentNode)
+	livekitServer, err := NewLivekitServer(conf, roomService, egressService, ingressService, sipService, ioInfoService, rtcService, agentService, keyProvider, router, roomManager, signalServer, server, currentNode)
 	if err != nil {
 		return nil, err
 	}
@@ -237,6 +244,19 @@ func getIngressConfig(conf *config.Config) *config.IngressConfig {
 	return &conf.Ingress
 }
 
+func getSIPStore(s ObjectStore) SIPStore {
+	switch store := s.(type) {
+	case *RedisStore:
+		return store
+	default:
+		return nil
+	}
+}
+
+func getSIPConfig(conf *config.Config) *config.SIPConfig {
+	return &conf.SIP
+}
+
 func createClientConfiguration() clientconfiguration.ClientConfigurationManager {
 	return clientconfiguration.NewStaticClientConfigurationManager(clientconfiguration.StaticConfigurations)
 }