**Add webhook event publishing for music state changes**

Introduced webhook event publishing to push music-state updates to a configured endpoint. Implemented `WebhookDispatcher` and `NoopEventPublisher` for dispatching events asynchronously or disabling them when no URL is set. Added handling of events like track start/end, queue updates, volume changes, and voice connection changes, enhancing integration with external systems.
This commit is contained in:
anthorob committed 2026-10-26 11:43:07 -04:00
1 parent 7443cbfb4b
commit b0664e8835
15 files changed
+513 -4

No files matched your search

+12 -2
View File
@@ -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"));
}, () -> { }));
@@ -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);
}
@@ -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);
}
}
@@ -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
}
}
@@ -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<MusicEvent> queue;
private final Map<String, AtomicLong> 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);
}
}
}
@@ -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<String, Object> 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.
@@ -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() {
@@ -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<String, MusicManager> managers = new ConcurrentHashMap<>();
private final List<Disposable> 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)
);
}
@@ -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));
}
@@ -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<String, Object> 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());
@@ -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> 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<String> 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));
}
}