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.
This commit is contained in:
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user