diff --git a/.env.example b/.env.example index cc48acf..95223af 100644 --- a/.env.example +++ b/.env.example @@ -25,6 +25,8 @@ NETNEUTRALITY_WEBPO_URL=http://webpo:8080 NETNEUTRALITY_WEBPO_PASSWORD= NETNEUTRALITY_SENTRY_DSN= +NETNEUTRALITY_WEBHOOK_URL= +NETNEUTRALITY_WEBHOOK_SECRET= GITEA_HOST= GITEA_TOKEN= GITEA_OWNER= diff --git a/README.md b/README.md index c3e4959..4dddce7 100644 --- a/README.md +++ b/README.md @@ -37,6 +37,17 @@ Key variables: | `NETNEUTRALITY_LAVALINK_URI`, `NETNEUTRALITY_LAVALINK_PASSWORD` | Lavalink connection settings | | `NETNEUTRALITY_SENTRY_DSN` | Optional Sentry DSN for error reporting | | `GITEA_HOST`, `GITEA_TOKEN`, `GITEA_OWNER`, `GITEA_REPO` | Optional Gitea settings for issue reporting | +| `NETNEUTRALITY_WEBHOOK_URL`, `NETNEUTRALITY_WEBHOOK_SECRET` | Optional frontend webhook URL and HMAC secret (see below) | + +### Webhook events + +When `NETNEUTRALITY_WEBHOOK_URL` is set, the bot POSTs JSON music-state events (`TRACK_START`, `TRACK_END`, `QUEUE_UPDATED`, `PLAYER_PAUSED`, `PLAYER_RESUMED`, `VOLUME_CHANGED`, `VOICE_CONNECTED`, `VOICE_DISCONNECTED`) to it asynchronously. Each event has `type`, `guildId`, `timestamp`, a per-guild `seq` and a `payload`. Requests carry an `X-Signature: sha256=` header computed with `NETNEUTRALITY_WEBHOOK_SECRET`. Failed deliveries are retried 3 times with exponential backoff and then dropped, so the frontend should still do a full resync on load and periodically. The polling endpoints remain available. + +`PLAYER_SEEKED` is sent after a seek with payload `{ position }` (milliseconds). `VOTE_SKIP_UPDATED` is sent after a vote that doesn't skip the track yet, with payload `{ userId, votes, threshold }` (`threshold` is the percentage of listeners needed). A vote that does skip produces the usual `TRACK_END` / `TRACK_START` / `QUEUE_UPDATED` events instead. + +`TRACK_START` / `TRACK_END` payloads are `{ identifier, title, author, length, uri, artworkUrl }` (plus `reason` on `TRACK_END`). `uri` and `artworkUrl` may be `null` when the source doesn't provide them. + +`VOICE_CONNECTED` / `VOICE_DISCONNECTED` are sent for the bot itself and for any member joining or leaving the voice channel the bot is currently in. Their payload is `{ channelId, userId, bot, self }`, where `bot` marks bot accounts and `self` marks this bot's own voice changes. ## Running with Docker Compose diff --git a/bot.properties.example b/bot.properties.example index fcf31c8..8233f0f 100644 --- a/bot.properties.example +++ b/bot.properties.example @@ -25,6 +25,8 @@ NETNEUTRALITY_LAVALINK_PASSWORD=youshallnotpass NETNEUTRALITY_LAVALINK_NAME=main NETNEUTRALITY_SENTRY_DSN= +NETNEUTRALITY_WEBHOOK_URL= +NETNEUTRALITY_WEBHOOK_SECRET= GITEA_HOST= GITEA_TOKEN= GITEA_OWNER= diff --git a/pom.xml b/pom.xml index 5a63573..0934511 100644 --- a/pom.xml +++ b/pom.xml @@ -177,6 +177,7 @@ 5.3.2 + io.javalin javalin diff --git a/src/main/java/me/bobert/netneutrality/App.java b/src/main/java/me/bobert/netneutrality/App.java index e93f39e..f3e4dab 100644 --- a/src/main/java/me/bobert/netneutrality/App.java +++ b/src/main/java/me/bobert/netneutrality/App.java @@ -14,6 +14,9 @@ import me.bobert.netneutrality.bootstrap.BotState; import me.bobert.netneutrality.core.LogHandler; import me.bobert.netneutrality.core.command.CommandManager; import me.bobert.netneutrality.db.SessionManager; +import me.bobert.netneutrality.events.EventPublisher; +import me.bobert.netneutrality.events.NoopEventPublisher; +import me.bobert.netneutrality.events.WebhookDispatcher; import me.bobert.netneutrality.core.tasks.TaskManagerImpl; import me.bobert.netneutrality.core.scheduled.ScheduledTaskManager; import me.bobert.netneutrality.core.thread.ThreadManager; @@ -52,6 +55,7 @@ public class App { private final Settings settings; private JDA jda; private MusicService music; + private EventPublisher events = NoopEventPublisher.INSTANCE; private final ApplicationContext context = new ApplicationContext(); static void main() { @@ -146,8 +150,14 @@ public class App { UserFunction.bind(guildServices); DisabledCommandPrecondition.bind(guildServices); + // Webhook event dispatch (priority 50); stays a no-op publisher when no URL is configured. + String webhookUrl = settings.get("NETNEUTRALITY_WEBHOOK_URL"); + if (webhookUrl != null && !webhookUrl.isBlank()) { + events = context.register(new WebhookDispatcher(webhookUrl, settings.get("NETNEUTRALITY_WEBHOOK_SECRET"))); + } + // Music sources (priority 40): initialise the shared player manager; disconnect on shutdown. - music = context.register(new MusicServiceImpl(settings, guildServices, tasks, executors)); + music = context.register(new MusicServiceImpl(settings, guildServices, tasks, executors, events)); BotState.bind(guildServices, music); // JDA gateway; disconnected on shutdown. @@ -156,7 +166,7 @@ public class App { // Register slash commands, attach the event handler and set the presence once JDA is ready. context.register(lifecycle(65, () -> { commands.registerSlashCommands(jda); - jda.addEventListener(new EventsHandler(settings, guildServices, commands, tasks)); + jda.addEventListener(new EventsHandler(settings, guildServices, commands, tasks, events)); jda.getPresence().setActivity(Activity.watching("you")); }, () -> { })); diff --git a/src/main/java/me/bobert/netneutrality/events/EventPublisher.java b/src/main/java/me/bobert/netneutrality/events/EventPublisher.java new file mode 100644 index 0000000..88b1cd7 --- /dev/null +++ b/src/main/java/me/bobert/netneutrality/events/EventPublisher.java @@ -0,0 +1,15 @@ +package me.bobert.netneutrality.events; + +/* + * Copyright (c) anthorob + * + * https://discord.gg/qUa6zNFZsd + */ + +/** + * Sink for {@link MusicEvent}s. Implementations must never block the calling thread. + */ +public interface EventPublisher { + + void publish(MusicEvent event); +} diff --git a/src/main/java/me/bobert/netneutrality/events/MusicEvent.java b/src/main/java/me/bobert/netneutrality/events/MusicEvent.java new file mode 100644 index 0000000..f71b46b --- /dev/null +++ b/src/main/java/me/bobert/netneutrality/events/MusicEvent.java @@ -0,0 +1,39 @@ +package me.bobert.netneutrality.events; + +/* + * Copyright (c) anthorob + * + * https://discord.gg/qUa6zNFZsd + */ + +/** + * A music-state change pushed to the frontend webhook. + * + * @param type the kind of change + * @param guildId the guild the change belongs to + * @param timestamp epoch millis when the event was created + * @param seq per-guild, monotonically increasing sequence number (assigned by the dispatcher) + * @param payload event specific data, serialised as JSON + */ +public record MusicEvent(Type type, String guildId, long timestamp, long seq, Object payload) { + + public enum Type { + TRACK_START, + TRACK_END, + QUEUE_UPDATED, + PLAYER_PAUSED, + PLAYER_RESUMED, + VOLUME_CHANGED, + PLAYER_SEEKED, + VOTE_SKIP_UPDATED, + VOICE_CONNECTED, + VOICE_DISCONNECTED + } + + /** + * Creates an unsequenced event stamped with the current time. + */ + public static MusicEvent of(Type type, String guildId, Object payload) { + return new MusicEvent(type, guildId, System.currentTimeMillis(), 0L, payload); + } +} diff --git a/src/main/java/me/bobert/netneutrality/events/NoopEventPublisher.java b/src/main/java/me/bobert/netneutrality/events/NoopEventPublisher.java new file mode 100644 index 0000000..108b483 --- /dev/null +++ b/src/main/java/me/bobert/netneutrality/events/NoopEventPublisher.java @@ -0,0 +1,20 @@ +package me.bobert.netneutrality.events; + +/* + * Copyright (c) anthorob + * + * https://discord.gg/qUa6zNFZsd + */ + +/** + * Publisher that discards every event; used when webhooks are disabled. + */ +public final class NoopEventPublisher implements EventPublisher { + + public static final NoopEventPublisher INSTANCE = new NoopEventPublisher(); + + @Override + public void publish(MusicEvent event) { + // intentionally empty + } +} diff --git a/src/main/java/me/bobert/netneutrality/events/WebhookDispatcher.java b/src/main/java/me/bobert/netneutrality/events/WebhookDispatcher.java new file mode 100644 index 0000000..00e24fa --- /dev/null +++ b/src/main/java/me/bobert/netneutrality/events/WebhookDispatcher.java @@ -0,0 +1,166 @@ +package me.bobert.netneutrality.events; + +/* + * Copyright (c) anthorob + * + * https://discord.gg/qUa6zNFZsd + */ + +import com.fasterxml.jackson.databind.ObjectMapper; +import me.bobert.netneutrality.bootstrap.Lifecycle; +import okhttp3.MediaType; +import okhttp3.OkHttpClient; +import okhttp3.Request; +import okhttp3.RequestBody; +import okhttp3.Response; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import javax.crypto.Mac; +import javax.crypto.spec.SecretKeySpec; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.HexFormat; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.LinkedBlockingDeque; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +/** + * Delivers {@link MusicEvent}s to a single static webhook URL. Events are queued in a bounded + * buffer (oldest dropped on overflow) and sent in order by one worker thread, signed with + * HMAC-SHA256 in the {@code X-Signature} header. Failed deliveries are retried with exponential + * backoff and then dropped. + */ +public class WebhookDispatcher implements EventPublisher, Lifecycle { + + private static final Logger logger = LogManager.getLogger(WebhookDispatcher.class); + private static final MediaType JSON = MediaType.get("application/json; charset=utf-8"); + private static final String HMAC = "HmacSHA256"; + + private final String url; + private final String secret; + private final OkHttpClient client; + private final ObjectMapper mapper = new ObjectMapper(); + private final int maxAttempts; + private final long baseBackoffMillis; + private final LinkedBlockingDeque queue; + private final Map sequences = new ConcurrentHashMap<>(); + private volatile boolean running; + private Thread worker; + + public WebhookDispatcher(String url, String secret) { + this(url, secret, new OkHttpClient.Builder() + .callTimeout(Duration.ofSeconds(10)) + .build(), 1024, 3, 500); + } + + public WebhookDispatcher(String url, String secret, OkHttpClient client, int capacity, + int retries, long baseBackoffMillis) { + this.url = url; + this.secret = secret == null ? "" : secret; + this.client = client; + this.queue = new LinkedBlockingDeque<>(capacity); + this.maxAttempts = retries + 1; + this.baseBackoffMillis = baseBackoffMillis; + } + + @Override + public int priority() { + return 50; + } + + @Override + public void start() { + running = true; + worker = new Thread(this::run, "webhook-dispatcher"); + worker.setDaemon(true); + worker.start(); + logger.info("Webhook dispatcher enabled, posting to {}", url); + } + + @Override + public void stop() { + running = false; + if (worker != null) { + worker.interrupt(); + } + } + + @Override + public void publish(MusicEvent event) { + if (event == null) { + return; + } + long seq = sequences.computeIfAbsent(event.guildId(), k -> new AtomicLong()).incrementAndGet(); + MusicEvent sequenced = new MusicEvent(event.type(), event.guildId(), event.timestamp(), seq, event.payload()); + while (!queue.offerLast(sequenced)) { + MusicEvent dropped = queue.pollFirst(); + if (dropped != null) { + logger.warn("Webhook queue full, dropped {} for guild {} (seq {})", + dropped.type(), dropped.guildId(), dropped.seq()); + } + } + } + + private void run() { + while (running) { + try { + MusicEvent event = queue.pollFirst(1, TimeUnit.SECONDS); + if (event != null) { + deliver(event); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } catch (Exception e) { + logger.error("Unexpected webhook dispatch error", e); + } + } + } + + private void deliver(MusicEvent event) throws InterruptedException { + byte[] body; + try { + body = mapper.writeValueAsBytes(event); + } catch (Exception e) { + logger.error("Failed to serialise {} event, dropping", event.type(), e); + return; + } + + Request request = new Request.Builder() + .url(url) + .header("X-Signature", "sha256=" + sign(body)) + .post(RequestBody.create(body, JSON)) + .build(); + + for (int attempt = 1; attempt <= maxAttempts; attempt++) { + try (Response response = client.newCall(request).execute()) { + if (response.isSuccessful()) { + return; + } + logger.warn("Webhook returned {} for {} (attempt {}/{})", + response.code(), event.type(), attempt, maxAttempts); + } catch (Exception e) { + logger.warn("Webhook delivery failed for {} (attempt {}/{}): {}", + event.type(), attempt, maxAttempts, e.getMessage()); + } + if (attempt < maxAttempts) { + Thread.sleep(baseBackoffMillis << (attempt - 1)); + } + } + logger.error("Dropping {} event for guild {} (seq {}) after {} attempts", + event.type(), event.guildId(), event.seq(), maxAttempts); + } + + String sign(byte[] body) { + try { + Mac mac = Mac.getInstance(HMAC); + mac.init(new SecretKeySpec(secret.getBytes(StandardCharsets.UTF_8), HMAC)); + return HexFormat.of().formatHex(mac.doFinal(body)); + } catch (Exception e) { + throw new IllegalStateException("Unable to sign webhook payload", e); + } + } +} diff --git a/src/main/java/me/bobert/netneutrality/hooks/EventsHandler.java b/src/main/java/me/bobert/netneutrality/hooks/EventsHandler.java index e0472b0..c03907e 100644 --- a/src/main/java/me/bobert/netneutrality/hooks/EventsHandler.java +++ b/src/main/java/me/bobert/netneutrality/hooks/EventsHandler.java @@ -25,11 +25,17 @@ import me.bobert.netneutrality.core.command.CommandRegistry; import me.bobert.netneutrality.core.tasks.TaskManager; import me.bobert.netneutrality.db.GSettings; import me.bobert.netneutrality.utils.SettingsProvider; +import net.dv8tion.jda.api.entities.GuildVoiceState; import net.dv8tion.jda.api.entities.Message; +import net.dv8tion.jda.api.entities.channel.unions.AudioChannelUnion; import net.dv8tion.jda.api.events.Event; import net.dv8tion.jda.api.events.GenericEvent; import net.dv8tion.jda.api.events.guild.GuildJoinEvent; import net.dv8tion.jda.api.events.guild.GuildLeaveEvent; +import net.dv8tion.jda.api.events.guild.voice.GuildVoiceUpdateEvent; +import me.bobert.netneutrality.events.EventPublisher; +import me.bobert.netneutrality.events.MusicEvent; +import me.bobert.netneutrality.events.NoopEventPublisher; import net.dv8tion.jda.api.events.guild.update.GuildUpdateIconEvent; import net.dv8tion.jda.api.events.guild.update.GuildUpdateNameEvent; import net.dv8tion.jda.api.events.interaction.ModalInteractionEvent; @@ -58,6 +64,7 @@ public class EventsHandler implements EventListener { private final GuildServiceProvider guildServices; private final CommandRegistry commands; private final TaskManager taskManager; + private final EventPublisher events; public EventsHandler( SettingsProvider settings, @@ -65,6 +72,17 @@ public class EventsHandler implements EventListener { CommandRegistry commands, TaskManager taskManager ) { + this(settings, guildServices, commands, taskManager, NoopEventPublisher.INSTANCE); + } + + public EventsHandler( + SettingsProvider settings, + GuildServiceProvider guildServices, + CommandRegistry commands, + TaskManager taskManager, + EventPublisher events + ) { + this.events = Objects.requireNonNull(events, "events"); this.settings = Objects.requireNonNull(settings, "settings"); this.guildServices = Objects.requireNonNull(guildServices, "guildServices"); this.commands = Objects.requireNonNull(commands, "commands"); @@ -112,12 +130,51 @@ public class EventsHandler implements EventListener { } }); addGenericListener(GuildLeaveEvent.class, event -> onGuildLeave(event.getGuild().getId())); + addGenericListener(GuildVoiceUpdateEvent.class, this::onVoiceUpdate); addGenericListener(GuildUpdateNameEvent.class, event -> guildServices.get(event.getGuild().getId()).updateMetadata(event.getNewName(), event.getGuild().getIconId())); addGenericListener(GuildUpdateIconEvent.class, event -> guildServices.get(event.getGuild().getId()).updateMetadata(event.getGuild().getName(), event.getNewIconId())); } + /** + * Publishes connect/disconnect events for the bot itself and for members entering or leaving + * the voice channel the bot is currently in. The payload carries {@code userId}, {@code bot} + * (whether the member is a bot) and {@code self} (whether it is this bot). + */ + void onVoiceUpdate(GuildVoiceUpdateEvent event) { + boolean self = event.getEntity().getId().equals(event.getJDA().getSelfUser().getId()); + AudioChannelUnion left = event.getChannelLeft(); + AudioChannelUnion joined = event.getChannelJoined(); + + String botChannelId = null; + if (!self) { + GuildVoiceState botState = event.getGuild().getSelfMember().getVoiceState(); + if (botState == null || botState.getChannel() == null) { + return; + } + botChannelId = botState.getChannel().getId(); + } + + String guildId = event.getGuild().getId(); + if (left != null && (self || left.getId().equals(botChannelId))) { + events.publish(MusicEvent.of(MusicEvent.Type.VOICE_DISCONNECTED, guildId, + voicePayload(event, left.getId(), self))); + } + if (joined != null && (self || joined.getId().equals(botChannelId))) { + events.publish(MusicEvent.of(MusicEvent.Type.VOICE_CONNECTED, guildId, + voicePayload(event, joined.getId(), self))); + } + } + + private static Map voicePayload(GuildVoiceUpdateEvent event, String channelId, boolean self) { + return Map.of( + "channelId", channelId, + "userId", event.getEntity().getId(), + "bot", event.getEntity().getUser().isBot(), + "self", self); + } + /** * Deletes the guild's data and drops its cached service. The service is evicted even when the * delete fails, so a stale cache entry never outlives the guild membership. diff --git a/src/main/java/me/bobert/netneutrality/music/MusicManager.java b/src/main/java/me/bobert/netneutrality/music/MusicManager.java index aff422f..d4a4b91 100644 --- a/src/main/java/me/bobert/netneutrality/music/MusicManager.java +++ b/src/main/java/me/bobert/netneutrality/music/MusicManager.java @@ -24,6 +24,10 @@ import net.dv8tion.jda.api.entities.channel.middleman.AudioChannel; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; +import me.bobert.netneutrality.events.EventPublisher; +import me.bobert.netneutrality.events.MusicEvent; +import me.bobert.netneutrality.events.NoopEventPublisher; + import java.util.concurrent.CompletableFuture; import java.util.function.Consumer; @@ -49,8 +53,15 @@ public class MusicManager implements MusicManagerCallback { private final GuildService guildService; private final TaskManager taskManager; private final VoteSkipService voteSkipService; + private final EventPublisher events; MusicManager(Guild guild, LavalinkClient client, GuildService guildService, TaskManager taskManager) { + this(guild, client, guildService, taskManager, NoopEventPublisher.INSTANCE); + } + + MusicManager(Guild guild, LavalinkClient client, GuildService guildService, TaskManager taskManager, + EventPublisher events) { + this.events = events; this.guild = guild; this.client = client; this.scheduler = new TrackScheduler(this); @@ -88,6 +99,10 @@ public class MusicManager implements MusicManagerCallback { return voteSkipService; } + public EventPublisher getEvents() { + return events; + } + public Guild getGuild() { return guild; } @@ -324,6 +339,7 @@ public class MusicManager implements MusicManagerCallback { */ public synchronized void pauseTrack() { updatePlayer("pause track", builder -> builder.setPaused(true)); + events.publish(MusicEvent.of(MusicEvent.Type.PLAYER_PAUSED, guild.getId(), null)); } /** @@ -331,6 +347,7 @@ public class MusicManager implements MusicManagerCallback { */ public synchronized void unpauseTrack() { updatePlayer("unpause track", builder -> builder.setPaused(false)); + events.publish(MusicEvent.of(MusicEvent.Type.PLAYER_RESUMED, guild.getId(), null)); } /** @@ -353,7 +370,15 @@ public class MusicManager implements MusicManagerCallback { */ public synchronized QueueActionResult voteSkip(Member member) { int threshold = parseVoteSkipThreshold(guildService); - return voteSkipService.voteSkip(member, guild.getId(), getQueueManager(), getConnectedChannel(), isCurrentlyPlaying(), threshold); + TrackInfo before = getQueueManager().peekHeadOfQueue(); + QueueActionResult result = voteSkipService.voteSkip(member, guild.getId(), getQueueManager(), getConnectedChannel(), isCurrentlyPlaying(), threshold); + if (before != null && before == getQueueManager().peekHeadOfQueue()) { + events.publish(MusicEvent.of(MusicEvent.Type.VOTE_SKIP_UPDATED, guild.getId(), java.util.Map.of( + "userId", member.getId(), + "votes", before.getSkipCount(), + "threshold", threshold))); + } + return result; } /** @@ -373,6 +398,7 @@ public class MusicManager implements MusicManagerCallback { public void setVolume(int volume) { updatePlayer("set volume", builder -> builder.setVolume(volume)); + events.publish(MusicEvent.of(MusicEvent.Type.VOLUME_CHANGED, guild.getId(), java.util.Map.of("volume", volume))); } /** @@ -394,6 +420,7 @@ public class MusicManager implements MusicManagerCallback { */ public void seek(long positionMs) { updatePlayer("seek to " + positionMs, builder -> builder.setPosition(positionMs)); + events.publish(MusicEvent.of(MusicEvent.Type.PLAYER_SEEKED, guild.getId(), java.util.Map.of("position", positionMs))); } public int getQueueSize() { diff --git a/src/main/java/me/bobert/netneutrality/music/MusicServiceImpl.java b/src/main/java/me/bobert/netneutrality/music/MusicServiceImpl.java index f149150..035816f 100644 --- a/src/main/java/me/bobert/netneutrality/music/MusicServiceImpl.java +++ b/src/main/java/me/bobert/netneutrality/music/MusicServiceImpl.java @@ -34,6 +34,8 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; +import me.bobert.netneutrality.events.EventPublisher; +import me.bobert.netneutrality.events.NoopEventPublisher; /** * Instance-based {@link MusicService} owning the shared {@link LavalinkClient} and the @@ -54,6 +56,7 @@ public class MusicServiceImpl implements MusicService { private final GuildServiceProvider guildServices; private final TaskManager taskManager; private final ExecutorProvider executors; + private final EventPublisher events; private final Map managers = new ConcurrentHashMap<>(); private final List subscriptions = new ArrayList<>(); private final LavalinkClient client; @@ -61,6 +64,12 @@ public class MusicServiceImpl implements MusicService { public MusicServiceImpl(SettingsProvider settings, GuildServiceProvider guildServices, TaskManager taskManager, ExecutorProvider executors) { + this(settings, guildServices, taskManager, executors, NoopEventPublisher.INSTANCE); + } + + public MusicServiceImpl(SettingsProvider settings, GuildServiceProvider guildServices, TaskManager taskManager, + ExecutorProvider executors, EventPublisher events) { + this.events = events; this.settings = settings; this.guildServices = guildServices; this.taskManager = taskManager; @@ -105,7 +114,7 @@ public class MusicServiceImpl implements MusicService { public MusicManager get(Guild guild) { return managers.computeIfAbsent( guild.getId(), - id -> new MusicManager(guild, client, guildServices.get(id), taskManager) + id -> new MusicManager(guild, client, guildServices.get(id), taskManager, events) ); } diff --git a/src/main/java/me/bobert/netneutrality/music/QueueManager.java b/src/main/java/me/bobert/netneutrality/music/QueueManager.java index ac3b48d..d2524d0 100644 --- a/src/main/java/me/bobert/netneutrality/music/QueueManager.java +++ b/src/main/java/me/bobert/netneutrality/music/QueueManager.java @@ -6,6 +6,7 @@ package me.bobert.netneutrality.music; */ import dev.arbjerg.lavalink.client.player.Track; +import me.bobert.netneutrality.events.MusicEvent; import net.dv8tion.jda.api.entities.Guild; import net.dv8tion.jda.api.entities.Member; import org.apache.logging.log4j.LogManager; @@ -15,6 +16,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.Deque; import java.util.List; +import java.util.Map; import java.util.concurrent.LinkedBlockingDeque; import java.util.function.Predicate; @@ -62,9 +64,19 @@ public class QueueManager { public void removeHeadOfQueue() { synchronized (queue) { queue.poll(); + notifyQueueUpdated(); } } + private void notifyQueueUpdated() { + MusicManager musicManager = scheduler.getMusicManager(); + if (musicManager == null || musicManager.getEvents() == null) { + return; + } + musicManager.getEvents().publish(MusicEvent.of(MusicEvent.Type.QUEUE_UPDATED, guild.getId(), + Map.of("size", queue.size()))); + } + public boolean isEmpty() { synchronized (queue) { return queue.isEmpty(); @@ -79,6 +91,7 @@ public class QueueManager { } queue.clear(); + notifyQueueUpdated(); scheduler.getCallback().onQueueEmpty(); return QueueActionResult.success(String.format("Removed %d track%s from the queue", queueSize, (queueSize > 1 ? "s" : ""))); } @@ -97,6 +110,7 @@ public class QueueManager { TrackInfo first = queue.peek(); queue.removeIf(trackInfo -> member.equals(trackInfo.getRequester())); + notifyQueueUpdated(); if (first != null && member.equals(first.getRequester())) { skip(1); @@ -134,6 +148,7 @@ public class QueueManager { } else { queue.add(info); } + notifyQueueUpdated(); if (idle) { TrackInfo head = queue.peek(); @@ -182,6 +197,7 @@ public class QueueManager { public void startNextTrack() { synchronized (queue) { queue.poll(); + notifyQueueUpdated(); TrackInfo next = queue.peek(); if (next != null) { @@ -243,6 +259,7 @@ public class QueueManager { queue.clear(); queue.add(current); queue.addAll(newQueue); + notifyQueueUpdated(); return QueueActionResult.success("The queue is now shuffled."); } @@ -266,6 +283,7 @@ public class QueueManager { queue.clear(); queue.addAll(newQueue); + notifyQueueUpdated(); return QueueActionResult.success(String.format("Removed track #%d from the queue", index)); } @@ -309,6 +327,7 @@ public class QueueManager { queue.clear(); queue.addAll(newQueue); + notifyQueueUpdated(); return QueueActionResult.success(String.format("Moved track #%d to position #%d", fromIndex, toIndex)); } diff --git a/src/main/java/me/bobert/netneutrality/music/TrackScheduler.java b/src/main/java/me/bobert/netneutrality/music/TrackScheduler.java index 450f120..2ea77de 100644 --- a/src/main/java/me/bobert/netneutrality/music/TrackScheduler.java +++ b/src/main/java/me/bobert/netneutrality/music/TrackScheduler.java @@ -11,6 +11,7 @@ import dev.arbjerg.lavalink.protocol.v4.Message.EmittedEvent.TrackEndEvent; import me.bobert.netneutrality.context.MusicChannelContext; import me.bobert.netneutrality.core.LogHandler; import me.bobert.netneutrality.core.context.Context; +import me.bobert.netneutrality.events.MusicEvent; import me.bobert.netneutrality.exceptions.ContextException; import net.dv8tion.jda.api.entities.Guild; import net.dv8tion.jda.api.entities.Member; @@ -102,6 +103,7 @@ public class TrackScheduler { logger.debug("Guild {}: starting track: {}", guild.getId(), track.getInfo().getIdentifier()); currentTrack = track; stuckTrackRecoveryHandler.reset(); + publish(MusicEvent.Type.TRACK_START, track, null); } public void onTrackEnd(Track track, TrackEndEvent.AudioTrackEndReason endReason) { @@ -110,6 +112,7 @@ public class TrackScheduler { if (endReason != TrackEndEvent.AudioTrackEndReason.REPLACED) { currentTrack = null; } + publish(MusicEvent.Type.TRACK_END, track, endReason.name()); if (endReason == TrackEndEvent.AudioTrackEndReason.LOAD_FAILED) { logger.debug("Guild {}: failed loading track: {}", guild.getId(), track.getInfo().getIdentifier()); @@ -134,6 +137,23 @@ public class TrackScheduler { // clone of this back to your queue } + private void publish(MusicEvent.Type type, Track track, String reason) { + if (manager.getEvents() == null) { + return; + } + java.util.Map payload = new java.util.HashMap<>(); + payload.put("identifier", track.getInfo().getIdentifier()); + payload.put("title", track.getInfo().getTitle()); + payload.put("author", track.getInfo().getAuthor()); + payload.put("length", track.getInfo().getLength()); + payload.put("uri", track.getInfo().getUri()); + payload.put("artworkUrl", track.getInfo().getArtworkUrl()); + if (reason != null) { + payload.put("reason", reason); + } + manager.getEvents().publish(MusicEvent.of(type, guild.getId(), payload)); + } + public void onTrackException(Track track, TrackException exception) { logger.error("Guild {}: exception while playing track: {}", guild.getId(), track.getInfo().getIdentifier()); diff --git a/src/test/java/me/bobert/netneutrality/events/WebhookDispatcherTest.java b/src/test/java/me/bobert/netneutrality/events/WebhookDispatcherTest.java new file mode 100644 index 0000000..4edf333 --- /dev/null +++ b/src/test/java/me/bobert/netneutrality/events/WebhookDispatcherTest.java @@ -0,0 +1,111 @@ +package me.bobert.netneutrality.events; + +import com.sun.net.httpserver.HttpServer; +import okhttp3.OkHttpClient; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class WebhookDispatcherTest { + + private record Received(String signature, String body) { + } + + private HttpServer server; + private WebhookDispatcher dispatcher; + private final BlockingQueue received = new LinkedBlockingQueue<>(); + private final AtomicInteger failuresLeft = new AtomicInteger(); + + @BeforeEach + void setup() throws Exception { + server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/hook", exchange -> { + String body = new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8); + received.add(new Received(exchange.getRequestHeaders().getFirst("X-Signature"), body)); + int code = failuresLeft.getAndUpdate(n -> Math.max(0, n - 1)) > 0 ? 500 : 200; + exchange.sendResponseHeaders(code, -1); + exchange.close(); + }); + server.start(); + dispatcher = new WebhookDispatcher("http://127.0.0.1:" + server.getAddress().getPort() + "/hook", + "s3cret", new OkHttpClient(), 16, 2, 10); + dispatcher.start(); + } + + @AfterEach + void teardown() { + dispatcher.stop(); + server.stop(0); + } + + private Received next() throws InterruptedException { + return received.poll(5, TimeUnit.SECONDS); + } + + @Test + void shouldSignAndDeliverEvent() throws Exception { + dispatcher.publish(MusicEvent.of(MusicEvent.Type.TRACK_START, "1", null)); + + Received r = next(); + assertNotNull(r); + assertEquals("sha256=" + dispatcher.sign(r.body().getBytes(StandardCharsets.UTF_8)), r.signature()); + assertTrue(r.body().contains("TRACK_START")); + } + + @Test + void shouldRetryThenSucceed() throws Exception { + failuresLeft.set(1); + + dispatcher.publish(MusicEvent.of(MusicEvent.Type.QUEUE_UPDATED, "1", null)); + + assertNotNull(next()); + assertNotNull(next()); + assertNull(received.poll(300, TimeUnit.MILLISECONDS)); + } + + @Test + void shouldDropAfterRetriesExhausted() throws Exception { + failuresLeft.set(100); + + dispatcher.publish(MusicEvent.of(MusicEvent.Type.VOLUME_CHANGED, "1", null)); + + for (int i = 0; i < 3; i++) { + assertNotNull(next()); + } + assertNull(received.poll(300, TimeUnit.MILLISECONDS)); + } + + @Test + void shouldAssignPerGuildSequenceInOrder() throws Exception { + dispatcher.publish(MusicEvent.of(MusicEvent.Type.TRACK_START, "1", null)); + dispatcher.publish(MusicEvent.of(MusicEvent.Type.TRACK_END, "1", null)); + dispatcher.publish(MusicEvent.of(MusicEvent.Type.TRACK_START, "2", null)); + + List expected = List.of("\"seq\":1", "\"seq\":2", "\"seq\":1"); + for (String seq : expected) { + Received r = next(); + assertNotNull(r); + assertTrue(r.body().contains(seq), r.body()); + } + } + + @Test + void noopPublisherShouldMakeNoCalls() throws Exception { + NoopEventPublisher.INSTANCE.publish(MusicEvent.of(MusicEvent.Type.TRACK_START, "1", null)); + + assertNull(received.poll(200, TimeUnit.MILLISECONDS)); + } +}