module.go

package module

import (
	"context"
	"crypto/sha256"
	"encoding/base32"
	"fmt"
	"net"
	"os"
	"path/filepath"
	"sync"
	"time"

	"github.com/fullstorydev/grpcurl"
	grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
	"github.com/jhump/protoreflect/desc"
	"github.com/jhump/protoreflect/grpcreflect"
	"github.com/pion/rtp"
	"github.com/pion/webrtc/v3"
	"github.com/pkg/errors"
	"go.opencensus.io/trace"
	"go.uber.org/multierr"
	pb "go.viam.com/api/module/v1"
	robotpb "go.viam.com/api/robot/v1"
	streampb "go.viam.com/api/stream/v1"
	"go.viam.com/utils"
	"go.viam.com/utils/rpc"
	"golang.org/x/exp/maps"
	"google.golang.org/grpc"
	reflectpb "google.golang.org/grpc/reflection/grpc_reflection_v1alpha"

	"go.viam.com/rdk/components/camera/rtppassthrough"
	"go.viam.com/rdk/config"
	"go.viam.com/rdk/logging"
	"go.viam.com/rdk/operation"
	"go.viam.com/rdk/protoutils"
	"go.viam.com/rdk/resource"
	"go.viam.com/rdk/robot/client"
	rutils "go.viam.com/rdk/utils"
)

const (
	socketSuffix = ".sock"
	// socketHashSuffixLength determines how many characters from the module's name's hash should be used when truncating the module socket.
	socketHashSuffixLength int = 5
	// maxSocketAddressLength is the length (-1 for null terminator) of the .sun_path field as used in kernel bind()/connect() syscalls.
	// Linux allows for a max length of 107 but to simplify this code, we truncate to the macOS limit of 103.
	socketMaxAddressLength int = 103
	rtpBufferSize          int = 512
)

// CreateSocketAddress returns a socket address of the form parentDir/desiredName.sock
// if it is shorter than the socketMaxAddressLength. If this path would be too long, this function
// truncates desiredName and returns parentDir/truncatedName-hashOfDesiredName.sock.
//
// Importantly, this function will return the same socket address as long as the desiredName doesn't change.
func CreateSocketAddress(parentDir, desiredName string) (string, error) {
	baseAddr := filepath.ToSlash(parentDir)
	numRemainingChars := socketMaxAddressLength -
		len(baseAddr) -
		len(socketSuffix) -
		1 // `/` between baseAddr and name
	if numRemainingChars < len(desiredName) && numRemainingChars < socketHashSuffixLength+1 {
		return "", errors.Errorf("module socket base path would result in a path greater than the OS limit of %d characters: %s",
			socketMaxAddressLength, baseAddr)
	}
	// If possible, early-exit with a non-truncated socket path
	if numRemainingChars >= len(desiredName) {
		return filepath.Join(baseAddr, desiredName+socketSuffix), nil
	}
	// Hash the desiredName so that every invocation returns the same truncated address
	desiredNameHashCreator := sha256.New()
	_, err := desiredNameHashCreator.Write([]byte(desiredName))
	if err != nil {
		return "", errors.Errorf("failed to calculate a hash for %q while creating a truncated socket address", desiredName)
	}
	desiredNameHash := base32.StdEncoding.EncodeToString(desiredNameHashCreator.Sum(nil))
	if len(desiredNameHash) < socketHashSuffixLength {
		// sha256.Sum() should return 32 bytes so this shouldn't occur, but good to check instead of panicing
		return "", errors.Errorf("the encoded hash %q for %q is shorter than the minimum socket suffix length %v",
			desiredNameHash, desiredName, socketHashSuffixLength)
	}
	// Assemble the truncated socket address
	socketHashSuffix := desiredNameHash[:socketHashSuffixLength]
	truncatedName := desiredName[:(numRemainingChars - socketHashSuffixLength - 1)]
	return filepath.Join(baseAddr, fmt.Sprintf("%s-%s%s", truncatedName, socketHashSuffix, socketSuffix)), nil
}

// HandlerMap is the format for api->model pairs that the module will service.
// Ex: mymap["rdk:component:motor"] = ["acme:marine:thruster", "acme:marine:outboard"].
type HandlerMap map[resource.RPCAPI][]resource.Model

// ToProto converts the HandlerMap to a protobuf representation.
func (h HandlerMap) ToProto() *pb.HandlerMap {
	pMap := &pb.HandlerMap{}
	for s, models := range h {
		subtype := &robotpb.ResourceRPCSubtype{
			Subtype: protoutils.ResourceNameToProto(resource.Name{
				API:  s.API,
				Name: "",
			}),
			ProtoService: s.ProtoSvcName,
		}

		handler := &pb.HandlerDefinition{Subtype: subtype}
		for _, m := range models {
			handler.Models = append(handler.Models, m.String())
		}
		pMap.Handlers = append(pMap.Handlers, handler)
	}
	return pMap
}

// NewHandlerMapFromProto converts protobuf to HandlerMap.
func NewHandlerMapFromProto(ctx context.Context, pMap *pb.HandlerMap, conn rpc.ClientConn) (HandlerMap, error) {
	hMap := make(HandlerMap)
	refClient := grpcreflect.NewClientV1Alpha(ctx, reflectpb.NewServerReflectionClient(conn))
	defer refClient.Reset()
	reflSource := grpcurl.DescriptorSourceFromServer(ctx, refClient)

	var errs error
	for _, h := range pMap.GetHandlers() {
		api := protoutils.ResourceNameFromProto(h.Subtype.Subtype).API

		symDesc, err := reflSource.FindSymbol(h.Subtype.ProtoService)
		if err != nil {
			errs = multierr.Combine(errs, err)
			if errors.Is(err, grpcurl.ErrReflectionNotSupported) {
				return nil, errs
			}
			continue
		}
		svcDesc, ok := symDesc.(*desc.ServiceDescriptor)
		if !ok {
			return nil, errors.Errorf("expected descriptor to be service descriptor but got %T", symDesc)
		}
		rpcAPI := &resource.RPCAPI{
			API:  api,
			Desc: svcDesc,
		}
		for _, m := range h.Models {
			model, err := resource.NewModelFromString(m)
			if err != nil {
				return nil, err
			}
			hMap[*rpcAPI] = append(hMap[*rpcAPI], model)
		}
	}
	return hMap, errs
}

type peerResourceState struct {
	// NOTE As I'm only suppporting video to start this will always be a single element
	// once we add audio we will need to make this a slice / map
	subID rtppassthrough.SubscriptionID
	// NOTE As I'm only suppporting video to start this will always be a single element
	// once we add audio we will need to make this a slice
	sender *webrtc.RTPSender
}

// Module represents an external resource module that services components/services.
type Module struct {
	parent                  *client.RobotClient
	server                  rpc.Server
	logger                  logging.Logger
	mu                      sync.Mutex
	activeResourceStreams   map[resource.Name]peerResourceState
	streamSourceByName      map[resource.Name]rtppassthrough.Source
	operations              *operation.Manager
	ready                   bool
	addr                    string
	parentAddr              string
	activeBackgroundWorkers sync.WaitGroup
	handlers                HandlerMap
	collections             map[resource.API]resource.APIResourceCollection[resource.Resource]
	resLoggers              map[resource.Resource]logging.Logger
	closeOnce               sync.Once
	pc                      *webrtc.PeerConnection
	pcReady                 <-chan struct{}
	pcFailed                <-chan struct{}
	pb.UnimplementedModuleServiceServer
	streampb.UnimplementedStreamServiceServer
}

// NewModule returns the basic module framework/structure.
func NewModule(ctx context.Context, address string, logger logging.Logger) (*Module, error) {
	// TODO(PRODUCT-343): session support likely means interceptors here
	opMgr := operation.NewManager(logger)
	unaries := []grpc.UnaryServerInterceptor{
		opMgr.UnaryServerInterceptor,
	}
	streams := []grpc.StreamServerInterceptor{
		opMgr.StreamServerInterceptor,
	}
	opts := []grpc.ServerOption{
		grpc.UnaryInterceptor(grpc_middleware.ChainUnaryServer(unaries...)),
		grpc.StreamInterceptor(grpc_middleware.ChainStreamServer(streams...)),
	}
	m := &Module{
		logger:                logger,
		addr:                  address,
		operations:            opMgr,
		streamSourceByName:    map[resource.Name]rtppassthrough.Source{},
		activeResourceStreams: map[resource.Name]peerResourceState{},
		server:                NewServer(opts...),
		ready:                 true,
		handlers:              HandlerMap{},
		collections:           map[resource.API]resource.APIResourceCollection[resource.Resource]{},
		resLoggers:            map[resource.Resource]logging.Logger{},
	}
	if err := m.server.RegisterServiceServer(ctx, &pb.ModuleService_ServiceDesc, m); err != nil {
		return nil, err
	}
	if err := m.server.RegisterServiceServer(ctx, &streampb.StreamService_ServiceDesc, m); err != nil {
		return nil, err
	}

	// attempt to construct a PeerConnection
	pc, err := webrtc.NewPeerConnection(webrtc.Configuration{})
	if err != nil {
		logger.Debugw("Unable to create optional peer connection for module. Skipping WebRTC for module...", "err", err)
		return m, nil
	}

	// attempt to configure PeerConnection
	pcReady, err := rpc.ConfigureForRenegotiation(pc, logger.AsZap())
	if err != nil {
		msg := "Error creating renegotiation channel for module. Unable to " +
			"create optional peer connection for module. Skipping WebRTC for module..."
		logger.Debugw(msg, "err", err)
		return m, nil
	}

	m.pc = pc
	m.pcReady = pcReady

	return m, nil
}

// NewModuleFromArgs directly parses the command line argument to get its address.
func NewModuleFromArgs(ctx context.Context, logger logging.Logger) (*Module, error) {
	if len(os.Args) < 2 {
		return nil, errors.New("need socket path as command line argument")
	}
	return NewModule(ctx, os.Args[1], logger)
}

// Start starts the module service and grpc server.
func (m *Module) Start(ctx context.Context) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	var lis net.Listener
	if err := MakeSelfOwnedFilesFunc(func() error {
		var err error
		lis, err = net.Listen("unix", m.addr)
		if err != nil {
			return errors.WithMessage(err, "failed to listen")
		}
		return nil
	}); err != nil {
		return err
	}

	m.activeBackgroundWorkers.Add(1)
	utils.PanicCapturingGo(func() {
		defer m.activeBackgroundWorkers.Done()
		// Attempt to remove module's .sock file.
		defer rutils.RemoveFileNoError(m.addr)
		m.logger.Infof("server listening at %v", lis.Addr())
		if err := m.server.Serve(lis); err != nil {
			m.logger.Errorf("failed to serve: %v", err)
		}
	})
	return nil
}

// Close shuts down the module and grpc server.
// TODO: Shut down the subscriptions.
func (m *Module) Close(ctx context.Context) {
	m.closeOnce.Do(func() {
		m.mu.Lock()
		parent := m.parent
		for name, r := range m.activeResourceStreams {
			vcss, ok := m.streamSourceByName[name]
			if !ok {
				m.logger.CError(ctx, "Module Close unable to find resource", "name", name,
					"in streamSourceByName", fmt.Sprintf("%#v", m.streamSourceByName))
				continue
			}
			if err := vcss.Unsubscribe(ctx, r.subID); err != nil {
				m.logger.CErrorw(ctx, "Unsubscribe", "name", name, "err", err, "subID", r.subID.String())
			}
			if err := m.pc.RemoveTrack(r.sender); err != nil {
				m.logger.CErrorw(ctx, "RemoveTrack", "name", name, "err", err)
			}
		}
		if m.pc != nil {
			if err := m.pc.Close(); err != nil {
				m.logger.CErrorw(ctx, "WebRTC Peer Connection Close", "err", err)
			}
		}
		m.mu.Unlock()
		m.logger.Info("Shutting down gracefully.")
		if parent != nil {
			if err := parent.Close(ctx); err != nil {
				m.logger.Error(err)
			}
		}
		if err := m.server.Stop(); err != nil {
			m.logger.Error(err)
		}
		m.activeBackgroundWorkers.Wait()
	})
}

// GetParentResource returns a resource from the parent robot by name.
func (m *Module) GetParentResource(ctx context.Context, name resource.Name) (resource.Resource, error) {
	// Refresh parent to ensure it has the most up-to-date resources before calling
	// ResourceByName.
	if err := m.parent.Refresh(ctx); err != nil {
		return nil, err
	}
	return m.parent.ResourceByName(name)
}

func (m *Module) connectParent(ctx context.Context) error {
	// If parent connection has already been made, do not make another one. Some
	// tests send two ReadyRequests sequentially, and if an rdk were to retry
	// sending a ReadyRequest to a module for any reason, we could feasibly make
	// a second connection back to the parent and leak the first, so disallow the
	// setting of parent more than once.
	if m.parent != nil {
		return nil
	}

	if err := CheckSocketOwner(m.parentAddr); err != nil {
		return err
	}

	// moduleLoggers may be creating the client connection below, so use a
	// different logger here to avoid a deadlock where the client connection
	// tries to recursively connect to the parent.
	clientLogger := logging.NewLogger("module-connection")
	clientLogger.SetLevel(m.logger.GetLevel())
	// TODO(PRODUCT-343): add session support to modules
	rc, err := client.New(ctx, "unix://"+m.parentAddr, clientLogger, client.WithDisableSessions())
	if err != nil {
		return err
	}

	m.parent = rc
	return nil
}

// SetReady can be set to false if the module is not ready (ex. waiting on hardware).
func (m *Module) SetReady(ready bool) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.ready = ready
}

// PeerConnect returns the encoded answer string for the `ReadyResponse`.
func (m *Module) PeerConnect(encodedOffer string) (string, error) {
	if m.pc == nil {
		return "", errors.New("no PeerConnection object")
	}

	offer := webrtc.SessionDescription{}
	if err := rpc.DecodeSDP(encodedOffer, &offer); err != nil {
		return "", err
	}
	if err := m.pc.SetRemoteDescription(offer); err != nil {
		return "", err
	}

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

	if err := m.pc.SetLocalDescription(answer); err != nil {
		return "", err
	}

	<-webrtc.GatheringCompletePromise(m.pc)
	return rpc.EncodeSDP(m.pc.LocalDescription())
}

// Ready receives the parent address and reports api/model combos the module is ready to service.
func (m *Module) Ready(ctx context.Context, req *pb.ReadyRequest) (*pb.ReadyResponse, error) {
	resp := &pb.ReadyResponse{}

	encodedAnswer, err := m.PeerConnect(req.WebrtcOffer)
	if err == nil {
		resp.WebrtcAnswer = encodedAnswer
	} else {
		m.logger.Debugw("Unable to create optional peer connection for module. Skipping WebRTC for module...", "err", err)
		pcFailed := make(chan struct{})
		close(pcFailed)
		m.pcFailed = pcFailed
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	m.parentAddr = req.GetParentAddress()
	if err := m.connectParent(ctx); err != nil {
		// Return error back to parent if we cannot make a connection from module
		// -> parent. Something is wrong in that case and the module should not be
		// operational.
		return nil, err
	}

	// If logger is a moduleLogger, start gRPC logging.
	if moduleLogger, ok := m.logger.(*moduleLogger); ok {
		moduleLogger.startLoggingViaGRPC(m)
	}

	resp.Ready = m.ready
	resp.Handlermap = m.handlers.ToProto()
	return resp, nil
}

// AddResource receives the component/service configuration from the parent.
func (m *Module) AddResource(ctx context.Context, req *pb.AddResourceRequest) (*pb.AddResourceResponse, error) {
	select {
	case <-m.pcReady:
	case <-m.pcFailed:
	}

	deps := make(resource.Dependencies)
	for _, c := range req.Dependencies {
		name, err := resource.NewFromString(c)
		if err != nil {
			return nil, err
		}
		c, err := m.GetParentResource(ctx, name)
		if err != nil {
			return nil, err
		}
		deps[name] = c
	}

	conf, err := config.ComponentConfigFromProto(req.Config)
	if err != nil {
		return nil, err
	}

	if err := addConvertedAttributes(conf); err != nil {
		return nil, errors.Wrapf(err, "unable to convert attributes when adding resource")
	}

	resInfo, ok := resource.LookupRegistration(conf.API, conf.Model)
	if !ok {
		return nil, errors.Errorf("do not know how to construct %q", conf.API)
	}
	if resInfo.Constructor == nil {
		return nil, errors.Errorf("invariant: no constructor for %q", conf.API)
	}
	resLogger := m.logger.Sublogger(conf.ResourceName().String())
	res, err := resInfo.Constructor(ctx, deps, *conf, resLogger)
	if err != nil {
		return nil, err
	}

	var passthroughSource rtppassthrough.Source
	if p, ok := res.(rtppassthrough.Source); ok {
		passthroughSource = p
	}

	m.mu.Lock()
	defer m.mu.Unlock()
	// add the video stream resources upon creation
	if passthroughSource != nil {
		m.streamSourceByName[res.Name()] = passthroughSource
	}
	coll, ok := m.collections[conf.API]
	if !ok {
		return nil, errors.Errorf("module cannot service api: %s", conf.API)
	}
	m.resLoggers[res] = resLogger

	return &pb.AddResourceResponse{}, coll.Add(conf.ResourceName(), res)
}

// ReconfigureResource receives the component/service configuration from the parent.
func (m *Module) ReconfigureResource(ctx context.Context, req *pb.ReconfigureResourceRequest) (*pb.ReconfigureResourceResponse, error) {
	var res resource.Resource
	deps := make(resource.Dependencies)
	for _, c := range req.Dependencies {
		name, err := resource.NewFromString(c)
		if err != nil {
			return nil, err
		}
		c, err := m.GetParentResource(ctx, name)
		if err != nil {
			return nil, err
		}
		deps[name] = c
	}

	// it is assumed the caller robot has handled model differences
	conf, err := config.ComponentConfigFromProto(req.Config)
	if err != nil {
		return nil, err
	}

	if err := addConvertedAttributes(conf); err != nil {
		return nil, errors.Wrapf(err, "unable to convert attributes when reconfiguring resource")
	}

	m.mu.Lock()
	defer m.mu.Unlock()

	coll, ok := m.collections[conf.API]
	if !ok {
		return nil, errors.Errorf("no rpc service for %+v", conf)
	}
	res, err = coll.Resource(conf.ResourceName().Name)
	if err != nil {
		return nil, err
	}

	if logger, ok := m.resLoggers[res]; ok {
		levelStr := req.GetConfig().GetLogConfiguration().GetLevel()
		if level, err := logging.LevelFromString(levelStr); err == nil {
			logger.SetLevel(level)
		} else {
			m.logger.Warnw("LogConfiguration does not contain a valid level.", "resource", res.Name().Name, "level", levelStr)
		}
	}

	reconfErr := res.Reconfigure(ctx, deps, *conf)
	if reconfErr == nil {
		return &pb.ReconfigureResourceResponse{}, nil
	}

	if !resource.IsMustRebuildError(reconfErr) {
		return nil, err
	}

	m.logger.Debugw("rebuilding", "name", conf.ResourceName())
	if err := res.Close(ctx); err != nil {
		m.logger.Error(err)
	}

	// check if our streams have been removed by reconfiguring and remove them if so
	prs, ok := m.activeResourceStreams[res.Name()]
	if ok {
		m.logger.CInfow(ctx, "Cleaning up orphaned activeResourceStreams for resource: ", "name", res.Name())
		if err := m.pc.RemoveTrack(prs.sender); err != nil {
			m.logger.Error(err)
		}

		delete(m.activeResourceStreams, res.Name())
	}

	resInfo, ok := resource.LookupRegistration(conf.API, conf.Model)
	if !ok {
		return nil, errors.Errorf("do not know how to construct %q", conf.API)
	}
	if resInfo.Constructor == nil {
		return nil, errors.Errorf("invariant: no constructor for %q", conf.API)
	}

	newRes, err := resInfo.Constructor(ctx, deps, *conf, m.logger)
	if err != nil {
		return nil, err
	}
	var passthroughSource rtppassthrough.Source
	if p, ok := newRes.(rtppassthrough.Source); ok {
		passthroughSource = p
	}

	if passthroughSource != nil {
		m.streamSourceByName[res.Name()] = passthroughSource
	}
	return &pb.ReconfigureResourceResponse{}, coll.ReplaceOne(conf.ResourceName(), newRes)
}

// ValidateConfig receives the validation request for a resource from the parent.
func (m *Module) ValidateConfig(ctx context.Context,
	req *pb.ValidateConfigRequest,
) (*pb.ValidateConfigResponse, error) {
	c, err := config.ComponentConfigFromProto(req.Config)
	if err != nil {
		return nil, err
	}

	if err := addConvertedAttributes(c); err != nil {
		return nil, errors.Wrapf(err, "unable to convert attributes for validation")
	}

	if c.ConvertedAttributes != nil {
		implicitDeps, err := c.ConvertedAttributes.Validate(c.Name)
		if err != nil {
			return nil, errors.Wrapf(err, "error validating resource")
		}
		return &pb.ValidateConfigResponse{Dependencies: implicitDeps}, nil
	}

	// Resource configuration object does not implement Validate, but return an
	// empty response and no error to maintain backward compatibility.
	return &pb.ValidateConfigResponse{}, nil
}

// RemoveResource receives the request for resource removal.
func (m *Module) RemoveResource(ctx context.Context, req *pb.RemoveResourceRequest) (*pb.RemoveResourceResponse, error) {
	slowWatcher, slowWatcherCancel := utils.SlowGoroutineWatcher(
		30*time.Second, fmt.Sprintf("module resource %q is taking a while to remove", req.Name), m.logger.AsZap())
	defer func() {
		slowWatcherCancel()
		<-slowWatcher
	}()
	m.mu.Lock()
	defer m.mu.Unlock()

	name, err := resource.NewFromString(req.Name)
	if err != nil {
		return nil, err
	}

	coll, ok := m.collections[name.API]
	if !ok {
		return nil, errors.Errorf("no grpc service for %+v", name)
	}
	res, err := coll.Resource(name.Name)
	if err != nil {
		return nil, err
	}

	// remove vidostreams when the camera resource implementing a passthrough is removed.
	vcss, streamSourceByNameOk := m.streamSourceByName[name]
	prs, activeResourceStreamsOk := m.activeResourceStreams[name]
	if streamSourceByNameOk && activeResourceStreamsOk {
		m.logger.CInfow(ctx, "calling RemoveTrack on", "name", name)
		if err := m.pc.RemoveTrack(prs.sender); err != nil {
			m.logger.CWarnw(ctx, "RemoveStream > RemoveTrack", "name", name, "err", err)
		}
		m.logger.CInfow(ctx, "calling Unsubscribe on", "name", name)
		if err := vcss.Unsubscribe(ctx, prs.subID); err != nil {
			m.logger.CWarnw(ctx, "RemoveStream > Unsubscribe", "name", name, "subID", prs.subID.String(), "err", err)
		}
		m.logger.CInfow(ctx, "deleting from activeResourceStreams", "name", name)
		delete(m.activeResourceStreams, name)
	}

	if err := res.Close(ctx); err != nil {
		m.logger.Error(err)
	}

	delete(m.streamSourceByName, res.Name())

	return &pb.RemoveResourceResponse{}, coll.Remove(name)
}

// addAPIFromRegistry adds a preregistered API (rpc API) to the module's services.
func (m *Module) addAPIFromRegistry(ctx context.Context, api resource.API) error {
	m.mu.Lock()
	defer m.mu.Unlock()
	_, ok := m.collections[api]
	if ok {
		return nil
	}

	apiInfo, ok := resource.LookupGenericAPIRegistration(api)
	if !ok {
		return errors.Errorf("invariant: registration does not exist for %q", api)
	}

	newColl := apiInfo.MakeEmptyCollection()
	m.collections[api] = newColl

	if !ok {
		return nil
	}
	return apiInfo.RegisterRPCService(ctx, m.server, newColl)
}

// AddModelFromRegistry adds a preregistered component or service model to the module's services.
func (m *Module) AddModelFromRegistry(ctx context.Context, api resource.API, model resource.Model) error {
	err := validateRegistered(api, model)
	if err != nil {
		return err
	}

	m.mu.Lock()
	_, ok := m.collections[api]
	m.mu.Unlock()
	if !ok {
		if err := m.addAPIFromRegistry(ctx, api); err != nil {
			return err
		}
	}

	apiInfo, ok := resource.LookupGenericAPIRegistration(api)
	if !ok {
		return errors.Errorf("invariant: registration does not exist for %q", api)
	}
	if apiInfo.ReflectRPCServiceDesc == nil {
		m.logger.Errorf("rpc subtype %s doesn't contain a valid ReflectRPCServiceDesc", api)
	}
	rpcAPI := resource.RPCAPI{
		API:          api,
		ProtoSvcName: apiInfo.RPCServiceDesc.ServiceName,
		Desc:         apiInfo.ReflectRPCServiceDesc,
	}

	m.mu.Lock()
	defer m.mu.Unlock()
	m.handlers[rpcAPI] = append(m.handlers[rpcAPI], model)
	return nil
}

// OperationManager returns the operation manager for the module.
func (m *Module) OperationManager() *operation.Manager {
	return m.operations
}

// ListStreams lists the streams.
func (m *Module) ListStreams(ctx context.Context, req *streampb.ListStreamsRequest) (*streampb.ListStreamsResponse, error) {
	_, span := trace.StartSpan(ctx, "module::module::ListStreams")
	defer span.End()
	names := make([]string, 0, len(m.streamSourceByName))
	for _, n := range maps.Keys(m.streamSourceByName) {
		names = append(names, n.String())
	}
	return &streampb.ListStreamsResponse{Names: names}, nil
}

// AddStream adds a stream.
func (m *Module) AddStream(ctx context.Context, req *streampb.AddStreamRequest) (*streampb.AddStreamResponse, error) {
	ctx, span := trace.StartSpan(ctx, "module::module::AddStream")
	defer span.End()
	name, err := resource.NewFromString(req.GetName())
	if err != nil {
		return nil, err
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.pc == nil {
		return nil, errors.New("module has no peer connection")
	}
	vcss, ok := m.streamSourceByName[name]
	if !ok {
		err := errors.New("unknown stream for resource")
		m.logger.CWarnw(ctx, err.Error(), "name", name, "streamSourceByName", fmt.Sprintf("%#v", m.streamSourceByName))
		return nil, err
	}

	if _, ok = m.activeResourceStreams[name]; ok {
		m.logger.CWarnw(ctx, "AddStream called with when there is already a stream for peer connection. NoOp", "name", name)
		return &streampb.AddStreamResponse{}, nil
	}

	tlsRTP, err := webrtc.NewTrackLocalStaticRTP(webrtc.RTPCodecCapability{MimeType: "video/H264"}, "video", name.String())
	if err != nil {
		return nil, errors.Wrap(err, "error creating a new TrackLocalStaticRTP")
	}

	subID, err := vcss.SubscribeRTP(ctx, rtpBufferSize, func(pkts []*rtp.Packet) error {
		for _, pkt := range pkts {
			if err := tlsRTP.WriteRTP(pkt); err != nil {
				m.logger.CWarnw(ctx, "SubscribeRTP callback function WriteRTP", "err", err)
			}
		}
		return nil
	})
	if err != nil {
		return nil, errors.Wrap(err, "error setting up stream subscription")
	}

	m.logger.CInfow(ctx, "AddStream calling AddTrack", "name", name, "tlsRTP", tlsRTP)
	sender, err := m.pc.AddTrack(tlsRTP)
	if err != nil {
		return nil, errors.Wrap(err, "error adding track")
	}

	m.activeResourceStreams[name] = peerResourceState{sender: sender, subID: subID}
	return &streampb.AddStreamResponse{}, nil
}

// RemoveStream removes a stream.
func (m *Module) RemoveStream(ctx context.Context, req *streampb.RemoveStreamRequest) (*streampb.RemoveStreamResponse, error) {
	ctx, span := trace.StartSpan(ctx, "module::module::RemoveStream")
	defer span.End()
	name, err := resource.NewFromString(req.GetName())
	if err != nil {
		return nil, err
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	if m.pc == nil {
		return nil, errors.New("module has no peer connection")
	}
	vcss, ok := m.streamSourceByName[name]
	if !ok {
		return nil, errors.Errorf("unknown stream for resource %s", name)
	}

	prs, ok := m.activeResourceStreams[name]
	if !ok {
		return nil, errors.Errorf("stream %s is not active", name)
	}

	if err := m.pc.RemoveTrack(prs.sender); err != nil {
		m.logger.CWarnw(ctx, "RemoveStream > RemoveTrack", "name", name, "subID", prs.subID.String(), "err", err)
	}

	if err := vcss.Unsubscribe(ctx, prs.subID); err != nil {
		m.logger.CWarnw(ctx, "RemoveStream > Unsubscribe", "name", name, "subID", prs.subID.String(), "err", err)
	}

	delete(m.activeResourceStreams, name)
	return &streampb.RemoveStreamResponse{}, nil
}

// addConvertedAttributesToConfig uses the MapAttributeConverter to fill in the
// ConvertedAttributes field from the Attributes and AssociatedResourceConfigs.
func addConvertedAttributes(cfg *resource.Config) error {
	// Try to find map converter for a resource.
	reg, ok := resource.LookupRegistration(cfg.API, cfg.Model)
	if ok && reg.AttributeMapConverter != nil {
		converted, err := reg.AttributeMapConverter(cfg.Attributes)
		if err != nil {
			return errors.Wrapf(err, "error converting attributes for resource")
		}
		cfg.ConvertedAttributes = converted
	}

	// Also try for associated configs (will only succeed if module itself registers the associated config API).
	for subIdx, associatedConf := range cfg.AssociatedResourceConfigs {
		conv, ok := resource.LookupAssociatedConfigRegistration(associatedConf.API)
		if !ok {
			continue
		}
		if conv.AttributeMapConverter != nil {
			converted, err := conv.AttributeMapConverter(associatedConf.Attributes)
			if err != nil {
				return errors.Wrap(err, "error converting associated resource config attributes")
			}
			// associated resource configs for resources might be missing a resource name
			// which can be inferred from its resource config.
			converted.UpdateResourceNames(func(oldName resource.Name) resource.Name {
				return cfg.ResourceName()
			})
			cfg.AssociatedResourceConfigs[subIdx].ConvertedAttributes = converted
		}
	}
	return nil
}

// validateRegistered returns an error if the passed-in api and model have not
// yet been registered.
func validateRegistered(api resource.API, model resource.Model) error {
	resInfo, ok := resource.LookupRegistration(api, model)
	if ok && resInfo.Constructor != nil {
		return nil
	}

	return errors.Errorf("resource with API %s and model %s not yet registered", api, model)
}