You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: handlers.go
    path of file: ./repos/streamdal/apps/server/services/bus
    
    the code of the file until where you have to start completion: package bus

import (
	"context"

	"github.com/pkg/errors"
	"github.com/sirupsen/logrus"

	"github.com/streamdal/streamdal/libs/protos/build/go/protos"

	"github.com/streamdal/streamdal/apps/server/services/store"
	stypes "github.com/streamdal/streamdal/apps/server/services/store/types"
	"github.com/streamdal/streamdal/apps/server/types"
	"github.com/streamdal/streamdal/apps/server/util"
	"github.com/streamdal/streamdal/apps/server/validate"
)

func (b *Bus) handleUpdatePipelineRequest(ctx context.Context, req *protos.UpdatePipelineRequest) error {
	llog := b.log.WithField("method", "handleUpdatePipelineRequest")
	llog.Debugf("handling update pipeline request bus event: %v", req.Pipeline.Name)

	if err := validate.UpdatePipelineRequest(req); err != nil {
		return errors.Wrap(err, "validation error")
	}

	// Inform the UI that changes have occurred
	b.op
    
func (b *Bus) handlePausePipelineRequest(ctx context.Context, req *protos.PausePipelineRequest) error {
	llog := b.log.WithField("method", "handlePausePipelineRequest")
	b.log.Debugf("handling pause pipeline request bus event: %v", req)

	// Inform the UI that changes have occurred
	b.options.PubSub.Publish(types.PubSubChangesTopic, "changes detected via pause pipeline broadcast handler")

	if err := validate.PausePipelineRequest(req); err != nil {
		return errors.Wrap(err, "validation error")
	}

	// Find all audiences + session ID's that use this pipeline ID
	usage, err := b.options.Store.GetActivePipelineUsage(ctx, req.PipelineId)
	if err != nil {
		llog.Errorf("error getting active pipeline usage for pipeline id '%s': %v", req.PipelineId, err)
		return errors.Wrapf(err, "error getting active pipeline usage for pipeline id '%s'", req.PipelineId)
	}

	// For each audience, generate pipeline config that EXCLUDE paused pipeline
	// and send the updated SetPipelines command to each session ID
	for _, u := range usage {
		pipelines, err := b.generatePipelinesForPauseResume(ctx, u.Audience)
		if err != nil {
			llog.Errorf("error generating pipelines for pause: %v", err)
			return errors.Wrap(err, "error generating pipelines for pause")
		}

		llog.Debugf("sending SetPipelines command to session id '%s' for audience '%s'",
			u.SessionId, util.AudienceToStr(u.Audience))

		if _, err := b.sendSetPipelinesCommand(u.Audience, pipelines, u.SessionId); err != nil {
			llog.Errorf("unable to send SetPipelines command: %v", err)
			return errors.Wrap(err, "error sending SetPipelines command")
		}
	}

	return nil
}