radio, rpc, config, mocks: add AnnouncerService.AnnounceUser

ircbot: implement the new AnnounceUser method that should be
called when the active stream user changes, this is done internally
with a StreamValue.

AnnounceUser will announce the user on the main irc channel with similar
output as what the .dj command normally returns with the new user info.
It will also try to change the topic after a small delay such that spam
updates won't cause a slew of topic updates.

StreamerUserInfo now only returns the current dj instead of also allowing
you to set it.

config/rpc.go

@@ -180,3 +180,8 @@ func (i *ircService) AnnounceRequest(ctx context.Context, song radio.Song) error
 func (i *ircService) AnnounceSong(ctx context.Context, status radio.Status) error {
 	return i.fn().AnnounceSong(ctx, status)
 }
+
+// AnnounceUser implements radio.AnnounceService.
+func (i *ircService) AnnounceUser(ctx context.Context, user *radio.User) error {
+	return i.fn().AnnounceUser(ctx, user)
+}

ircbot/api.go

@@ -2,8 +2,10 @@ package ircbot
 
 import (
 	"context"
+	"regexp"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	radio "github.com/R-a-dio/valkyrie"
@@ -37,6 +39,10 @@ type announceService struct {
 	bot                  *Bot
 	lastAnnounceSongTime time.Time
 	lastAnnounceSong     radio.Song
+
+	topicTimerMu  sync.Mutex
+	topicTimer    *time.Timer
+	topicLastEdit time.Time
 }
 
 func (ann *announceService) AnnounceSong(ctx context.Context, status radio.Status) error {
@@ -229,3 +235,88 @@ func (ann *announceService) AnnounceRequest(ctx context.Context, song radio.Song
 
 	return nil
 }
+
+func (ann *announceService) AnnounceUser(ctx context.Context, user *radio.User) error {
+	name := "None"
+	if user != nil {
+		name = user.DJ.Name
+	}
+
+	message := Fmt("Current DJ: {green}%s", name)
+	ann.bot.c.Cmd.Message(ann.Conf().IRC.MainChannel, message)
+
+	ann.queueChangeTopic(ctx, user)
+	return nil
+}
+
+func (ann *announceService) queueChangeTopic(ctx context.Context, user *radio.User) {
+	const topicDelay = time.Second * 15
+
+	ann.topicTimerMu.Lock()
+	if ann.topicTimer != nil {
+		// stop any timer that is already running
+		ann.topicTimer.Stop()
+	}
+	// start a timer for updating the topic
+	ann.topicTimer = time.AfterFunc(topicDelay, func() {
+		ann.topicTimerMu.Lock()
+		if time.Since(ann.topicLastEdit) < topicDelay {
+			// if our last edit was recent, just queue another update a bit
+			// into the future
+			ann.topicTimerMu.Unlock()
+			ann.queueChangeTopic(ctx, user)
+		}
+		defer ann.topicTimerMu.Unlock()
+
+		err := ann.changeTopic(context.WithoutCancel(ctx), user)
+		if err != nil {
+			zerolog.Ctx(ctx).Error().Err(err).Msg("failed to change topic")
+			return
+		}
+		ann.topicLastEdit = time.Now()
+		ann.topicTimer.Stop()
+		ann.topicTimer = nil
+	})
+	ann.topicTimerMu.Unlock()
+}
+
+var reOtherTopicBit = regexp.MustCompile(`(.*?r/.*/dio.*?)(\|.*?\|)(.*)`)
+
+func (ann *announceService) changeTopic(ctx context.Context, user *radio.User) error {
+	const op errors.Op = "ircbot/announceService.changeTopic"
+
+	channel := ann.bot.c.LookupChannel(ann.Conf().IRC.MainChannel)
+	if channel == nil {
+		return errors.E(op, "channel is missing")
+	}
+
+	// parse the topic so we can change it
+	match := reOtherTopicBit.FindStringSubmatch(channel.Topic)
+	if len(match) < 4 {
+		return errors.E(errors.BrokenTopic, op, errors.Info(channel.Topic))
+	}
+
+	topicStatus := "DOWN"
+	topicName := "None"
+	if user != nil {
+		topicStatus = "UP"
+		topicName = user.DJ.Name
+	}
+
+	// we get a []string back with all our groups, the first is the full match
+	// which we don't need
+	match = match[1:]
+	// now the group we're interested in is the second one, so replace that with
+	// our new status print
+	match[1] = Fmt(
+		"|{orange} Stream:{red} %s {orange}DJ:{red} %s {cyan} https://r-a-d.io {clear}|",
+		topicStatus, topicName,
+	)
+
+	newTopic := strings.Join(match, "")
+
+	if newTopic != channel.Topic {
+		ann.bot.c.Cmd.Topic(channel.Name, newTopic)
+	}
+	return nil
+}

ircbot/auth.go

@@ -27,7 +27,7 @@ func IsAuthed(e Event) bool {
 		select {
 		case authCh <- true:
 		default:
-			// this default cause should never happen, but it might be possible to
+			// this default case should never happen, but it might be possible to
 			// trigger it by having the same user call a command that checks
 			// authentication in rapid succession, so we protect against that
 		}
@@ -88,6 +88,7 @@ func HasAccess(c *girc.Client, e girc.Event) bool {
 // HasStreamAccess is similar to HasAccess but also includes special casing for streamers
 // that don't have channel access, but do have the authorization to access the stream
 func HasStreamAccess(c *girc.Client, e girc.Event) bool {
+	// TODO: implement this
 	return HasAccess(c, e)
 }
 

ircbot/commands_impl.go

@@ -88,8 +88,8 @@ func LastPlayed(e Event) error {
 func StreamerQueue(e Event) error {
 	const op errors.Op = "irc/StreamerQueue"
 
-	// Get queue from streamer
-	songQueue, err := e.Bot.Streamer.Queue(e.Ctx)
+	// Get queue
+	songQueue, err := e.Bot.Queue.Entries(e.Ctx)
 	if err != nil {
 		return errors.E(op, err)
 	}
@@ -189,61 +189,33 @@ func StreamerQueueLength(e Event) error {
 	return nil
 }
 
-var reOtherTopicBit = regexp.MustCompile(`(.*?r/.*/dio.*?)(\|.*?\|)(.*)`)
-
 func StreamerUserInfo(e Event) error {
 	const op errors.Op = "irc/StreamerUserInfo"
 
 	name := e.Arguments["DJ"]
 	if name == "" || !HasAccess(e.Client, e.Event) {
-		// simple path with no argument or no access
+		// simple path if people are just asking for the current dj
 		status := e.Bot.StatusValue.Latest()
 		e.EchoPublic("Current DJ: {green}%s", status.StreamerName)
 		return nil
 	}
 
-	channel := e.Client.LookupChannel(e.Params[0])
-	if channel == nil {
-		return nil
-	}
-
-	var err error
-	var user *radio.User
-	var topicStatus = "UP"
-
-	// skip the name lookup if the name is None, since it means we are down and out
-	if name != "None" {
-		user, err = e.Storage.User(e.Ctx).LookupName(name)
-		if err != nil {
-			return errors.E(op, err)
-		}
-	} else {
-		topicStatus = "DOWN"
-	}
-
-	err = e.Bot.Manager.UpdateUser(e.Ctx, user)
+	// lookup the name from the argument
+	user, err := e.Storage.User(e.Ctx).LookupName(name)
 	if err != nil {
 		return errors.E(op, err)
 	}
 
-	// parse the topic so we can change it
-	match := reOtherTopicBit.FindStringSubmatch(channel.Topic)
-	if len(match) < 4 {
-		return errors.E(errors.BrokenTopic, op, errors.Info(channel.Topic))
+	// user given isn't a robot, which means we're going to ignore it
+	if !radio.IsRobot(*user) {
+		e.EchoPublic("Current DJ: {green}%s", e.Bot.StatusValue.Latest().StreamerName)
+		return nil
 	}
 
-	// we get a []string back with all our groups, the first is the full match
-	// which we don't need
-	match = match[1:]
-	// now the group we're interested in is the second one, so replace that with
-	// our new status print
-	match[1] = Fmt(
-		"|{orange} Stream:{red} %s {orange}DJ:{red} %s {cyan} https://r-a-d.io {clear}|",
-		topicStatus, name,
-	)
+	// otherwise we should only care if the user is the one we are aware of
+	// but for now we only have one robot ever so just assume thats the value
 
-	newTopic := strings.Join(match, "")
-	e.Client.Cmd.Topic(channel.Name, newTopic)
+	// TODO: implement the poke to the streamer that tells it it can start
 	return nil
 }
 
@@ -421,7 +393,7 @@ func KillStreamer(e Event) error {
 	until := time.Until(status.SongInfo.End)
 	if force {
 		e.EchoPublic("Disconnecting right now")
-	} else if until == 0 {
+	} else if until <= 0 {
 		e.EchoPublic("Disconnecting after the current song")
 	} else {
 		e.EchoPublic("Disconnecting in about %s",

ircbot/main.go

@@ -47,11 +47,16 @@ func Execute(ctx context.Context, cfg config.Config) error {
 	b.StatusValue = util.StreamValue(ctx, cfg.Manager.CurrentStatus, func(ctx context.Context, s radio.Status) {
 		err := announce.AnnounceSong(ctx, s)
 		if err != nil {
-			zerolog.Ctx(ctx).Error().Err(err).Msg("failed to announce")
+			zerolog.Ctx(ctx).Error().Err(err).Msg("failed to announce status")
+		}
+	})
+	b.UserValue = util.StreamValue(ctx, cfg.Manager.CurrentUser, func(ctx context.Context, user *radio.User) {
+		err := announce.AnnounceUser(ctx, user)
+		if err != nil {
+			zerolog.Ctx(ctx).Error().Err(err).Msg("failed to announce user")
 		}
 	})
 	b.ListenersValue = util.StreamValue(ctx, cfg.Manager.CurrentListeners)
-	b.UserValue = util.StreamValue(ctx, cfg.Manager.CurrentUser)
 
 	errCh := make(chan error, 2)
 	go func() {
@@ -108,8 +113,6 @@ func NewBot(ctx context.Context, cfg config.Config) (*Bot, error) {
 	b := &Bot{
 		Config:   cfg,
 		Storage:  store,
-		Manager:  cfg.Manager,
-		Streamer: cfg.Streamer,
 		Searcher: ss,
 		c:        girc.New(ircConf),
 	}
@@ -130,8 +133,6 @@ type Bot struct {
 	Storage radio.StorageService
 
 	// interfaces to other components
-	Manager  radio.ManagerService
-	Streamer radio.StreamerService
 	Searcher radio.SearchService
 
 	// Values used by commands

radio.go

@@ -453,6 +453,7 @@ type QueueService interface {
 type AnnounceService interface {
 	AnnounceSong(context.Context, Status) error
 	AnnounceRequest(context.Context, Song) error
+	AnnounceUser(context.Context, *User) error
 }
 
 // SongID is a songs identifier

rpc/client.go

@@ -104,6 +104,15 @@ func (a AnnouncerClientRPC) AnnounceRequest(ctx context.Context, s radio.Song) e
 	return err
 }
 
+func (a AnnouncerClientRPC) AnnounceUser(ctx context.Context, u *radio.User) error {
+	ua := &UserAnnouncement{
+		User: toProtoUser(u),
+	}
+
+	_, err := a.rpc.AnnounceUser(ctx, ua)
+	return err
+}
+
 // NewManagerService returns a new client implementing radio.ManagerService
 func NewManagerService(c *grpc.ClientConn) radio.ManagerService {
 	return ManagerClientRPC{