From fcc7c9c66aee94fbbbb0e87236ac61581012b1ce Mon Sep 17 00:00:00 2001 From: FernandoJVideira <03.pleaser-minster@icloud.com> Date: Sat, 29 Aug 2026 19:22:16 +0100 Subject: [PATCH] feat(music): consume playback commands via LISTEN/NOTIFY The dashboard has been writing skip/pause/volume/etc. requests into music_commands with nothing on this end reading them. Adds a dedicated LISTEN connection (can't use the pooled *sql.DB for this, LISTEN has to stay bound to one specific connection) that wakes on NOTIFY, drains unprocessed rows, and dispatches each to the matching Manager function. A command's error is logged, not fatal, and every row gets marked processed regardless of outcome so a permanently broken command doesn't retry forever. --- internal/bot/bot.go | 11 +++- internal/music/listener.go | 106 +++++++++++++++++++++++++++++++++++++ main.go | 2 +- 3 files changed, 117 insertions(+), 2 deletions(-) create mode 100644 internal/music/listener.go diff --git a/internal/bot/bot.go b/internal/bot/bot.go index 1ba731b..cd2ad87 100644 --- a/internal/bot/bot.go +++ b/internal/bot/bot.go @@ -1,6 +1,7 @@ package bot import ( + "context" "log" "velox-bot/internal/commands" "velox-bot/internal/db/repos/musicrepo" @@ -19,11 +20,12 @@ type Bot struct { registeredCommands []*discordgo.ApplicationCommand Services *services.Services MusicRepo *musicrepo.Repo + DBHost string LavalinkHost string LavalinkPass string } -func NewBot(token, appID, guildID, lavalinkHost, lavalinkPass string, cmds []*discordgo.ApplicationCommand, services *services.Services, musicRepo *musicrepo.Repo) (*Bot, error) { +func NewBot(token, appID, guildID, lavalinkHost, lavalinkPass string, cmds []*discordgo.ApplicationCommand, services *services.Services, musicRepo *musicrepo.Repo, dbHost string) (*Bot, error) { session, err := discordgo.New("Bot " + token) if err != nil { return nil, err @@ -49,6 +51,7 @@ func NewBot(token, appID, guildID, lavalinkHost, lavalinkPass string, cmds []*di Commands: cmds, Services: services, MusicRepo: musicRepo, + DBHost: dbHost, LavalinkHost: lavalinkHost, LavalinkPass: lavalinkPass, }, nil @@ -61,6 +64,12 @@ func (b *Bot) Start() error { _ = music.Init(b.Session, b.MusicRepo, b.AppID, b.LavalinkHost, b.LavalinkPass) + go func() { + if err := music.StartCommandListener(context.Background(), b.DBHost, b.MusicRepo); err != nil { + log.Printf("music: command listener stopped: %v", err) + } + }() + b.registeredCommands = make([]*discordgo.ApplicationCommand, len(b.Commands)) for _, cmd := range b.Commands { created, err := b.Session.ApplicationCommandCreate(b.AppID, b.GuildID, cmd) diff --git a/internal/music/listener.go b/internal/music/listener.go new file mode 100644 index 0000000..b5ba307 --- /dev/null +++ b/internal/music/listener.go @@ -0,0 +1,106 @@ +package music + +import ( + "context" + "encoding/json" + "fmt" + "log" + "strconv" + "velox-bot/internal/db/repos/musicrepo" + + "github.com/jackc/pgx/v5" +) + +func StartCommandListener(ctx context.Context, connString string, repo *musicrepo.Repo) error { + conn, err := pgx.Connect(ctx, connString) + if err != nil { + return fmt.Errorf("connect: %w", err) + } + defer conn.Close(ctx) + + _, err = conn.Exec(ctx, "LISTEN music_commands") + if err != nil { + return fmt.Errorf("exec: %w", err) + } + + for { + _, err := conn.WaitForNotification(ctx) + if err != nil { + return fmt.Errorf("waitfornotification: %w", err) + } + + processCommands(ctx, repo) + } +} + +func processCommands(ctx context.Context, repo *musicrepo.Repo) { + cmds, err := repo.ListUnprocessedCommands(ctx) + if err != nil { + log.Printf("music: failed to list unprocessed commands: %v", err) + return + } + + for _, cmd := range cmds { + if err := dispatchCommand(cmd); err != nil { + log.Printf("music: command %d (%s) for guild %d failed: %v", cmd.ID, cmd.CommandType, cmd.GuildID, err) + } + + if err := repo.MarkCommandProcessed(ctx, cmd.ID); err != nil { + log.Printf("music: failed to mark command %d processed: %v", cmd.ID, err) + } + } +} + +func dispatchCommand(cmd musicrepo.MusicCommand) error { + guildID := strconv.FormatInt(cmd.GuildID, 10) + + switch cmd.CommandType { + case "skip": + return Skip(guildID) + + case "pause": + return Pause(guildID, true) + + case "resume": + return Pause(guildID, false) + + case "set_volume": + var payload struct { + Volume int `json:"volume"` + } + if err := json.Unmarshal(cmd.Payload, &payload); err != nil { + return fmt.Errorf("unmarshal payload: %w", err) + } + return SetVolume(guildID, payload.Volume) + + case "remove_track": + var payload struct { + QueueID string `json:"queue_id"` + } + if err := json.Unmarshal(cmd.Payload, &payload); err != nil { + return fmt.Errorf("unmarshal payload: %w", err) + } + return RemoveFromQueue(guildID, payload.QueueID) + + case "reorder": + var payload struct { + QueueID string `json:"queue_id"` + Position int `json:"position"` + } + if err := json.Unmarshal(cmd.Payload, &payload); err != nil { + return fmt.Errorf("unmarshal payload: %w", err) + } + return Reorder(guildID, payload.QueueID, payload.Position) + + case "repeat_song": + ToggleRepeatSong(guildID) + return nil + + case "repeat_queue": + ToggleRepeatQueue(guildID) + return nil + + default: + return fmt.Errorf("unknown command type %q", cmd.CommandType) + } +} diff --git a/main.go b/main.go index f246a65..a03d2c2 100644 --- a/main.go +++ b/main.go @@ -73,7 +73,7 @@ func main() { musicSettingsService := musicsettings.New(settingsRepo) services := services.NewServices(levelService, levelSettingsService, meetingService, scheduleService, userSettingsService, projectsService, rpsService, twitchService, welcomeService, defaultRoleService, logSettingsService, musicSettingsService) - bot, err := bot.NewBot(config.BotToken, config.AppID, config.GuildID, config.LavalinkHost, config.LavalinkPass, commands.AllCommands, services, musicRepo) + bot, err := bot.NewBot(config.BotToken, config.AppID, config.GuildID, config.LavalinkHost, config.LavalinkPass, commands.AllCommands, services, musicRepo, config.DBHost) if err != nil { log.Fatalf("Error creating bot: %v", err) return