feat: add Spigot event producer

src/main/java/games/dmg/spigotevents/SpigotEventProducerPlugin.java

@@ -0,0 +1,177 @@
+package games.dmg.spigotevents;
+
+import games.dmg.spigotevents.delivery.EventPipeline;
+import games.dmg.spigotevents.delivery.IngestClient;
+import games.dmg.spigotevents.delivery.OutboxDispatcher;
+import games.dmg.spigotevents.event.CloudEventFactory;
+import games.dmg.spigotevents.event.ServerIdentity;
+import games.dmg.spigotevents.listener.GameEventListener;
+import games.dmg.spigotevents.outbox.SqliteOutbox;
+import games.dmg.spigotevents.stats.PlayerSnapshotCollector;
+import games.dmg.spigotevents.stats.SnapshotScheduler;
+import java.net.URI;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.logging.Level;
+import org.bukkit.Bukkit;
+import org.bukkit.plugin.java.JavaPlugin;
+
+public final class SpigotEventProducerPlugin extends JavaPlugin {
+    private CloudEventFactory eventFactory;
+    private EventPipeline pipeline;
+    private SnapshotScheduler snapshotScheduler;
+
+    @Override
+    public void onEnable() {
+        saveDefaultConfig();
+        try {
+            PluginSettings settings = loadSettings();
+            String gameVersion = Bukkit.getBukkitVersion().split("-", 2)[0];
+            eventFactory = new CloudEventFactory(
+                    new ServerIdentity(settings.serverId(), settings.serverName()), gameVersion);
+
+            Path databasePath = getDataFolder().toPath().resolve("event-outbox.sqlite3");
+            SqliteOutbox outbox = new SqliteOutbox(databasePath);
+            var ingestClient = new IngestClient(settings.ingestUrl(), settings.httpTimeout());
+            var dispatcher = new OutboxDispatcher(
+                    outbox, ingestClient, settings.maxBatchEvents(), settings.maxBatchBytes());
+            pipeline = new EventPipeline(
+                    outbox,
+                    dispatcher,
+                    settings.sendInterval(),
+                    settings.dispatchThresholdEvents(),
+                    settings.dispatchThresholdBytes(),
+                    (message, error) -> getLogger().log(Level.WARNING, message, error));
+
+            getServer().getPluginManager().registerEvents(
+                    new GameEventListener(eventFactory, pipeline, Bukkit.getOnlinePlayers()), this);
+            var collector = new PlayerSnapshotCollector(
+                    eventFactory, pipeline, settings.statisticChunkBytes());
+            snapshotScheduler = new SnapshotScheduler(
+                    this,
+                    collector,
+                    settings.snapshotIntervalSeconds(),
+                    settings.statisticOperationsPerTick());
+            snapshotScheduler.start();
+
+            Map<String, Object> data = new LinkedHashMap<>();
+            data.put("plugin_version", getDescription().getVersion());
+            data.put("bukkit_version", Bukkit.getBukkitVersion());
+            data.put("server_implementation", Bukkit.getVersion());
+            pipeline.publish(eventFactory.create("server.started", data));
+            getLogger().info("Producing events as server '" + settings.serverName()
+                    + "' (" + settings.serverId() + ")");
+        } catch (Exception exception) {
+            getLogger().log(Level.SEVERE, "Could not initialize SpigotEventProducer", exception);
+            closeResources();
+            getServer().getPluginManager().disablePlugin(this);
+        }
+    }
+
+    @Override
+    public void onDisable() {
+        if (snapshotScheduler != null) {
+            snapshotScheduler.close();
+            snapshotScheduler = null;
+        }
+        if (pipeline != null && eventFactory != null) {
+            try {
+                pipeline.publishAndWait(
+                        eventFactory.create("server.stopped", Map.of()), Duration.ofSeconds(20));
+            } catch (Exception exception) {
+                getLogger().log(Level.WARNING, "Could not persist the server shutdown event", exception);
+            }
+        }
+        closeResources();
+    }
+
+    private PluginSettings loadSettings() {
+        String serverName = requireText("server-name");
+        String configuredId = getConfig().getString("server-id", "").trim();
+        UUID serverId;
+        if (configuredId.isEmpty()) {
+            serverId = UUID.randomUUID();
+            getConfig().set("server-id", serverId.toString());
+            saveConfig();
+        } else {
+            serverId = UUID.fromString(configuredId);
+        }
+
+        if (serverName.length() > 128) {
+            throw new IllegalArgumentException("server-name must be 128 characters or fewer");
+        }
+        int maxBatchBytes = positiveInt("batch.max-bytes");
+        int statisticChunkBytes = positiveInt("statistics.chunk-json-bytes");
+        if (statisticChunkBytes > maxBatchBytes - 8_192) {
+            throw new IllegalArgumentException(
+                    "statistics.chunk-json-bytes must leave at least 8192 bytes of batch headroom");
+        }
+
+        return new PluginSettings(
+                serverName,
+                serverId,
+                URI.create(requireText("ingest-url")),
+                Duration.ofSeconds(positiveLong("send-interval-seconds")),
+                positiveLong("snapshot-interval-seconds"),
+                Duration.ofSeconds(positiveLong("http-timeout-seconds")),
+                positiveInt("batch.max-events"),
+                maxBatchBytes,
+                positiveLong("dispatch-threshold.events"),
+                positiveLong("dispatch-threshold.bytes"),
+                statisticChunkBytes,
+                positiveInt("statistics.operations-per-tick"));
+    }
+
+    private String requireText(String path) {
+        String value = getConfig().getString(path, "").trim();
+        if (value.isEmpty()) {
+            throw new IllegalArgumentException(path + " must not be blank");
+        }
+        return value;
+    }
+
+    private int positiveInt(String path) {
+        int value = getConfig().getInt(path);
+        if (value < 1) {
+            throw new IllegalArgumentException(path + " must be positive");
+        }
+        return value;
+    }
+
+    private long positiveLong(String path) {
+        long value = getConfig().getLong(path);
+        if (value < 1) {
+            throw new IllegalArgumentException(path + " must be positive");
+        }
+        return value;
+    }
+
+    private void closeResources() {
+        if (pipeline != null) {
+            try {
+                pipeline.close();
+            } catch (Exception exception) {
+                getLogger().log(Level.WARNING, "Could not close the event outbox", exception);
+            }
+            pipeline = null;
+        }
+    }
+
+    private record PluginSettings(
+            String serverName,
+            UUID serverId,
+            URI ingestUrl,
+            Duration sendInterval,
+            long snapshotIntervalSeconds,
+            Duration httpTimeout,
+            int maxBatchEvents,
+            int maxBatchBytes,
+            long dispatchThresholdEvents,
+            long dispatchThresholdBytes,
+            int statisticChunkBytes,
+            int statisticOperationsPerTick) {
+    }
+}

src/main/java/games/dmg/spigotevents/chat/PrivateMessageParser.java

@@ -0,0 +1,77 @@
+package games.dmg.spigotevents.chat;
+
+import java.util.Locale;
+import java.util.Optional;
+import java.util.Set;
+
+public final class PrivateMessageParser {
+    private static final Set<String> EXPLICIT_COMMANDS = Set.of(
+            "msg", "message", "tell", "w", "whisper", "pm");
+    private static final Set<String> REPLY_COMMANDS = Set.of("reply", "r");
+
+    private PrivateMessageParser() {
+    }
+
+    public static Optional<PrivateMessage> parse(String commandLine) {
+        if (commandLine == null || commandLine.length() < 2 || commandLine.charAt(0) != '/') {
+            return Optional.empty();
+        }
+
+        int commandEnd = nextWhitespace(commandLine, 1);
+        if (commandEnd < 0) {
+            return Optional.empty();
+        }
+
+        String command = commandLine.substring(1, commandEnd).toLowerCase(Locale.ROOT);
+        int namespaceSeparator = command.lastIndexOf(':');
+        if (namespaceSeparator >= 0) {
+            command = command.substring(namespaceSeparator + 1);
+        }
+
+        int argumentStart = skipWhitespace(commandLine, commandEnd);
+        if (argumentStart >= commandLine.length()) {
+            return Optional.empty();
+        }
+
+        if (REPLY_COMMANDS.contains(command)) {
+            return Optional.of(new PrivateMessage(command, "", commandLine.substring(argumentStart)));
+        }
+        if (!EXPLICIT_COMMANDS.contains(command)) {
+            return Optional.empty();
+        }
+
+        int recipientEnd = nextWhitespace(commandLine, argumentStart);
+        if (recipientEnd < 0) {
+            return Optional.empty();
+        }
+        int messageStart = skipWhitespace(commandLine, recipientEnd);
+        if (messageStart >= commandLine.length()) {
+            return Optional.empty();
+        }
+
+        return Optional.of(new PrivateMessage(
+                command,
+                commandLine.substring(argumentStart, recipientEnd),
+                commandLine.substring(messageStart)));
+    }
+
+    private static int nextWhitespace(String value, int start) {
+        for (int index = start; index < value.length(); index++) {
+            if (Character.isWhitespace(value.charAt(index))) {
+                return index;
+            }
+        }
+        return -1;
+    }
+
+    private static int skipWhitespace(String value, int start) {
+        int index = start;
+        while (index < value.length() && Character.isWhitespace(value.charAt(index))) {
+            index++;
+        }
+        return index;
+    }
+
+    public record PrivateMessage(String command, String recipient, String message) {
+    }
+}

src/main/java/games/dmg/spigotevents/delivery/BatchSender.java

@@ -0,0 +1,9 @@
+package games.dmg.spigotevents.delivery;
+
+import games.dmg.spigotevents.outbox.StoredEvent;
+import java.util.List;
+
+@FunctionalInterface
+public interface BatchSender {
+    boolean send(List<StoredEvent> events) throws Exception;
+}

src/main/java/games/dmg/spigotevents/delivery/EventPipeline.java

@@ -0,0 +1,171 @@
+package games.dmg.spigotevents.delivery;
+
+import games.dmg.spigotevents.event.OutboundEvent;
+import games.dmg.spigotevents.outbox.SqliteOutbox;
+import java.time.Duration;
+import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.BiConsumer;
+
+public final class EventPipeline implements AutoCloseable {
+    private static final long INITIAL_RETRY_MILLIS = 5_000;
+
+    private final SqliteOutbox outbox;
+    private final OutboxDispatcher dispatcher;
+    private final long dispatchThresholdEvents;
+    private final long dispatchThresholdBytes;
+    private final long maxRetryMillis;
+    private final BiConsumer<String, Throwable> errorLogger;
+    private final ExecutorService persistenceWorker;
+    private final ScheduledThreadPoolExecutor deliveryWorker;
+    private final AtomicBoolean dispatchScheduled = new AtomicBoolean();
+    private final ScheduledFuture<?> periodicDispatch;
+    private volatile boolean closed;
+    private volatile long nextAttemptNanos;
+    private volatile long retryMillis = INITIAL_RETRY_MILLIS;
+
+    public EventPipeline(
+            SqliteOutbox outbox,
+            OutboxDispatcher dispatcher,
+            Duration sendInterval,
+            long dispatchThresholdEvents,
+            long dispatchThresholdBytes,
+            BiConsumer<String, Throwable> errorLogger) {
+        this.outbox = Objects.requireNonNull(outbox, "outbox");
+        this.dispatcher = Objects.requireNonNull(dispatcher, "dispatcher");
+        this.dispatchThresholdEvents = dispatchThresholdEvents;
+        this.dispatchThresholdBytes = dispatchThresholdBytes;
+        this.maxRetryMillis = Math.max(INITIAL_RETRY_MILLIS, sendInterval.toMillis());
+        this.errorLogger = Objects.requireNonNull(errorLogger, "errorLogger");
+        this.persistenceWorker = Executors.newSingleThreadExecutor(runnable -> daemonThread(
+                runnable, "spigot-event-persistence"));
+        this.deliveryWorker = new ScheduledThreadPoolExecutor(
+                1, runnable -> daemonThread(runnable, "spigot-event-delivery"));
+        deliveryWorker.setRemoveOnCancelPolicy(true);
+        long intervalMillis = Math.max(1, sendInterval.toMillis());
+        periodicDispatch = deliveryWorker.scheduleWithFixedDelay(
+                this::requestDispatch, 0, intervalMillis, TimeUnit.MILLISECONDS);
+    }
+
+    public void publish(OutboundEvent event) {
+        if (closed) {
+            errorLogger.accept("Rejected event after the pipeline closed: " + event.id(), null);
+            return;
+        }
+        try {
+            persistenceWorker.execute(() -> {
+                try {
+                    appendAndMaybeDispatch(event);
+                } catch (Exception exception) {
+                    errorLogger.accept("Could not persist event " + event.id(), exception);
+                }
+            });
+        } catch (RejectedExecutionException exception) {
+            errorLogger.accept("Rejected event while the pipeline was closing: " + event.id(), exception);
+        }
+    }
+
+    public void publishAndWait(OutboundEvent event, Duration timeout) throws Exception {
+        if (closed) {
+            throw new IllegalStateException("The event pipeline is closed");
+        }
+        persistenceWorker.submit(() -> {
+            appendAndMaybeDispatch(event);
+            return null;
+        }).get(timeout.toMillis(), TimeUnit.MILLISECONDS);
+    }
+
+    private void appendAndMaybeDispatch(OutboundEvent event) throws Exception {
+        outbox.append(event);
+        if (outbox.pendingCount() >= dispatchThresholdEvents
+                || outbox.pendingBytes() >= dispatchThresholdBytes) {
+            requestDispatch();
+        }
+    }
+
+    private void requestDispatch() {
+        if (closed || !dispatchScheduled.compareAndSet(false, true)) {
+            return;
+        }
+        long delayNanos = Math.max(0, nextAttemptNanos - System.nanoTime());
+        try {
+            deliveryWorker.schedule(this::dispatchAvailable, delayNanos, TimeUnit.NANOSECONDS);
+        } catch (RejectedExecutionException exception) {
+            dispatchScheduled.set(false);
+            if (!closed) {
+                errorLogger.accept("Could not schedule event delivery", exception);
+            }
+        }
+    }
+
+    private void dispatchAvailable() {
+        boolean failed = false;
+        try {
+            while (!closed && outbox.pendingCount() > 0) {
+                if (!dispatcher.dispatchOnce()) {
+                    failed = true;
+                    break;
+                }
+            }
+        } catch (Exception exception) {
+            failed = true;
+            errorLogger.accept("Could not deliver the event outbox", exception);
+        } finally {
+            if (failed) {
+                nextAttemptNanos = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(retryMillis);
+                retryMillis = Math.min(maxRetryMillis, retryMillis * 2);
+            } else {
+                nextAttemptNanos = 0;
+                retryMillis = INITIAL_RETRY_MILLIS;
+            }
+            dispatchScheduled.set(false);
+            if (failed && !closed) {
+                requestDispatch();
+            } else if (!closed) {
+                try {
+                    if (outbox.pendingCount() >= dispatchThresholdEvents
+                            || outbox.pendingBytes() >= dispatchThresholdBytes) {
+                        requestDispatch();
+                    }
+                } catch (Exception exception) {
+                    errorLogger.accept("Could not inspect the event outbox", exception);
+                }
+            }
+        }
+    }
+
+    @Override
+    public void close() throws Exception {
+        closed = true;
+        periodicDispatch.cancel(false);
+
+        persistenceWorker.shutdown();
+        boolean persistenceStopped = persistenceWorker.awaitTermination(20, TimeUnit.SECONDS);
+        if (!persistenceStopped) {
+            persistenceWorker.shutdownNow();
+            persistenceStopped = persistenceWorker.awaitTermination(5, TimeUnit.SECONDS);
+        }
+
+        deliveryWorker.shutdownNow();
+        boolean deliveryStopped = deliveryWorker.awaitTermination(20, TimeUnit.SECONDS);
+        if (!deliveryStopped) {
+            throw new IllegalStateException("Timed out while stopping event delivery");
+        }
+        if (!persistenceStopped) {
+            throw new IllegalStateException("Timed out while persisting queued events");
+        }
+        outbox.close();
+    }
+
+    private static Thread daemonThread(Runnable runnable, String name) {
+        Thread thread = new Thread(runnable, name);
+        thread.setDaemon(true);
+        return thread;
+    }
+}

src/main/java/games/dmg/spigotevents/delivery/IngestClient.java

@@ -0,0 +1,44 @@
+package games.dmg.spigotevents.delivery;
+
+import games.dmg.spigotevents.outbox.StoredEvent;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.time.Duration;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+public final class IngestClient implements BatchSender {
+    private final URI endpoint;
+    private final Duration requestTimeout;
+    private final HttpClient client;
+
+    public IngestClient(URI endpoint, Duration requestTimeout) {
+        this.endpoint = Objects.requireNonNull(endpoint, "endpoint");
+        this.requestTimeout = Objects.requireNonNull(requestTimeout, "requestTimeout");
+        this.client = HttpClient.newBuilder()
+                .connectTimeout(requestTimeout)
+                .followRedirects(HttpClient.Redirect.NORMAL)
+                .build();
+    }
+
+    @Override
+    public boolean send(List<StoredEvent> events) throws Exception {
+        if (events.isEmpty()) {
+            return true;
+        }
+        String body = events.stream()
+                .map(StoredEvent::payload)
+                .collect(Collectors.joining("\n", "", "\n"));
+        HttpRequest request = HttpRequest.newBuilder(endpoint)
+                .timeout(requestTimeout)
+                .header("Content-Type", "application/x-ndjson")
+                .header("Accept", "application/json, application/problem+json")
+                .POST(HttpRequest.BodyPublishers.ofString(body))
+                .build();
+        HttpResponse<Void> response = client.send(request, HttpResponse.BodyHandlers.discarding());
+        return response.statusCode() == 202;
+    }
+}

src/main/java/games/dmg/spigotevents/delivery/OutboxDispatcher.java

@@ -0,0 +1,30 @@
+package games.dmg.spigotevents.delivery;
+
+import games.dmg.spigotevents.outbox.SqliteOutbox;
+import java.util.Objects;
+
+public final class OutboxDispatcher {
+    private final SqliteOutbox outbox;
+    private final BatchSender sender;
+    private final int maxEvents;
+    private final int maxBytes;
+
+    public OutboxDispatcher(SqliteOutbox outbox, BatchSender sender, int maxEvents, int maxBytes) {
+        this.outbox = Objects.requireNonNull(outbox, "outbox");
+        this.sender = Objects.requireNonNull(sender, "sender");
+        this.maxEvents = maxEvents;
+        this.maxBytes = maxBytes;
+    }
+
+    public boolean dispatchOnce() throws Exception {
+        var events = outbox.peekBatch(maxEvents, maxBytes);
+        if (events.isEmpty()) {
+            return false;
+        }
+        if (!sender.send(events)) {
+            return false;
+        }
+        outbox.acknowledgeThrough(events.get(events.size() - 1).sequence());
+        return true;
+    }
+}

src/main/java/games/dmg/spigotevents/event/CloudEventFactory.java

@@ -0,0 +1,68 @@
+package games.dmg.spigotevents.event;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonObject;
+import java.time.Clock;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.UUID;
+import java.util.function.Supplier;
+
+public final class CloudEventFactory {
+    private static final Gson GSON = new Gson();
+    private static final String TYPE_PREFIX = "games.dmg.minecraft.";
+
+    private final ServerIdentity server;
+    private final String gameVersion;
+    private final Clock clock;
+    private final Supplier<UUID> eventIds;
+
+    public CloudEventFactory(ServerIdentity server, String gameVersion) {
+        this(server, gameVersion, Clock.systemUTC(), UUID::randomUUID);
+    }
+
+    CloudEventFactory(
+            ServerIdentity server,
+            String gameVersion,
+            Clock clock,
+            Supplier<UUID> eventIds) {
+        this.server = Objects.requireNonNull(server, "server");
+        this.gameVersion = Objects.requireNonNull(gameVersion, "gameVersion");
+        this.clock = Objects.requireNonNull(clock, "clock");
+        this.eventIds = Objects.requireNonNull(eventIds, "eventIds");
+    }
+
+    public OutboundEvent create(String eventName, PlayerIdentity player, Map<String, ?> eventData) {
+        UUID eventId = eventIds.get();
+        JsonObject event = new JsonObject();
+        event.addProperty("specversion", "1.0");
+        event.addProperty("id", eventId.toString());
+        event.addProperty("source", "urn:minecraft-server:" + server.id());
+        event.addProperty("type", TYPE_PREFIX + eventName);
+        event.addProperty("time", clock.instant().toString());
+        event.addProperty("datacontenttype", "application/json");
+        event.addProperty("serverid", server.id().toString());
+        event.addProperty("servername", server.name());
+        event.addProperty("gameversion", gameVersion);
+
+        Map<String, Object> data = new LinkedHashMap<>();
+        data.put("server_id", server.id().toString());
+        data.put("server_name", server.name());
+        data.put("game_version", gameVersion);
+        if (player != null) {
+            event.addProperty("playerid", player.id().toString());
+            data.put("player_id", player.id().toString());
+            data.put("player_name", player.name());
+        }
+        if (eventData != null) {
+            data.putAll(eventData);
+        }
+        event.add("data", GSON.toJsonTree(data));
+        return new OutboundEvent(eventId.toString(), GSON.toJson(event));
+    }
+
+    public OutboundEvent create(String eventName, Map<String, ?> eventData) {
+        return create(eventName, null, eventData);
+    }
+}

src/main/java/games/dmg/spigotevents/event/OutboundEvent.java

@@ -0,0 +1,15 @@
+package games.dmg.spigotevents.event;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Objects;
+
+public record OutboundEvent(String id, String payload) {
+    public OutboundEvent {
+        Objects.requireNonNull(id, "id");
+        Objects.requireNonNull(payload, "payload");
+    }
+
+    public int byteSize() {
+        return payload.getBytes(StandardCharsets.UTF_8).length + 1;
+    }
+}

src/main/java/games/dmg/spigotevents/event/PlayerIdentity.java

@@ -0,0 +1,11 @@
+package games.dmg.spigotevents.event;
+
+import java.util.Objects;
+import java.util.UUID;
+
+public record PlayerIdentity(UUID id, String name) {
+    public PlayerIdentity {
+        Objects.requireNonNull(id, "id");
+        Objects.requireNonNull(name, "name");
+    }
+}

src/main/java/games/dmg/spigotevents/event/ServerIdentity.java

@@ -0,0 +1,13 @@
+package games.dmg.spigotevents.event;
+
+import java.util.Objects;
+import java.util.UUID;
+
+public record ServerIdentity(UUID id, String name) {
+    public ServerIdentity {
+        Objects.requireNonNull(id, "id");
+        if (name == null || name.isBlank()) {
+            throw new IllegalArgumentException("Server name must not be blank");
+        }
+    }
+}

src/main/java/games/dmg/spigotevents/listener/GameEventListener.java

@@ -0,0 +1,83 @@
+package games.dmg.spigotevents.listener;
+
+import games.dmg.spigotevents.chat.PrivateMessageParser;
+import games.dmg.spigotevents.delivery.EventPipeline;
+import games.dmg.spigotevents.event.CloudEventFactory;
+import games.dmg.spigotevents.event.PlayerIdentity;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.IdentityHashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import org.bukkit.entity.Player;
+import org.bukkit.event.EventHandler;
+import org.bukkit.event.EventPriority;
+import org.bukkit.event.Listener;
+import org.bukkit.event.player.AsyncPlayerChatEvent;
+import org.bukkit.event.player.PlayerCommandPreprocessEvent;
+import org.bukkit.event.player.PlayerJoinEvent;
+import org.bukkit.event.player.PlayerQuitEvent;
+
+public final class GameEventListener implements Listener {
+    private final CloudEventFactory eventFactory;
+    private final EventPipeline pipeline;
+    private final Map<Player, PlayerIdentity> playerIdentities =
+            Collections.synchronizedMap(new IdentityHashMap<>());
+
+    public GameEventListener(
+            CloudEventFactory eventFactory,
+            EventPipeline pipeline,
+            Collection<? extends Player> onlinePlayers) {
+        this.eventFactory = eventFactory;
+        this.pipeline = pipeline;
+        for (Player player : onlinePlayers) {
+            playerIdentities.put(player, new PlayerIdentity(player.getUniqueId(), player.getName()));
+        }
+    }
+
+    @EventHandler(priority = EventPriority.MONITOR)
+    public void onJoin(PlayerJoinEvent event) {
+        Player player = event.getPlayer();
+        playerIdentities.put(player, new PlayerIdentity(player.getUniqueId(), player.getName()));
+        publishPlayer("player.joined", player, Map.of());
+    }
+
+    @EventHandler(priority = EventPriority.MONITOR)
+    public void onQuit(PlayerQuitEvent event) {
+        Player player = event.getPlayer();
+        publishPlayer("player.quit", player, Map.of());
+        playerIdentities.remove(player);
+    }
+
+    @SuppressWarnings("deprecation")
+    @EventHandler(priority = EventPriority.MONITOR, ignoreCancelled = true)
+    public void onChat(AsyncPlayerChatEvent event) {
+        publishPlayer("chat", event.getPlayer(), Map.of(
+                "message", event.getMessage(),
+                "asynchronous", event.isAsynchronous()));
+    }
+
+    @EventHandler(priority = EventPriority.MONITOR)
+    public void onCommand(PlayerCommandPreprocessEvent event) {
+        PrivateMessageParser.parse(event.getMessage()).ifPresent(message -> {
+            Map<String, Object> data = new LinkedHashMap<>();
+            data.put("message", message.message());
+            data.put("command", message.command());
+            data.put("reply", message.recipient().isEmpty());
+            if (!message.recipient().isEmpty()) {
+                data.put("recipient", message.recipient());
+            }
+            data.put("cancelled", event.isCancelled());
+            publishPlayer("private_message", event.getPlayer(), data);
+        });
+    }
+
+    private void publishPlayer(String type, Player player, Map<String, ?> data) {
+        PlayerIdentity identity = playerIdentities.get(player);
+        if (identity == null) {
+            identity = new PlayerIdentity(player.getUniqueId(), player.getName());
+            playerIdentities.put(player, identity);
+        }
+        pipeline.publish(eventFactory.create(type, identity, data));
+    }
+}

src/main/java/games/dmg/spigotevents/outbox/SqliteOutbox.java

@@ -0,0 +1,114 @@
+package games.dmg.spigotevents.outbox;
+
+import games.dmg.spigotevents.event.OutboundEvent;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+
+public final class SqliteOutbox implements AutoCloseable {
+    private final Connection connection;
+
+    public SqliteOutbox(Path databasePath) throws SQLException, IOException, ClassNotFoundException {
+        Path absolutePath = databasePath.toAbsolutePath();
+        Files.createDirectories(absolutePath.getParent());
+        Class.forName("org.sqlite.JDBC", true, SqliteOutbox.class.getClassLoader());
+        connection = DriverManager.getConnection("jdbc:sqlite:" + absolutePath);
+        initialize();
+    }
+
+    private void initialize() throws SQLException {
+        try (var statement = connection.createStatement()) {
+            statement.execute("PRAGMA journal_mode=WAL");
+            statement.execute("PRAGMA synchronous=FULL");
+            statement.execute("PRAGMA busy_timeout=5000");
+            statement.execute("""
+                    CREATE TABLE IF NOT EXISTS pending_events (
+                        sequence INTEGER PRIMARY KEY AUTOINCREMENT,
+                        event_id TEXT NOT NULL UNIQUE,
+                        payload TEXT NOT NULL,
+                        byte_size INTEGER NOT NULL,
+                        created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+                    )
+                    """);
+        }
+    }
+
+    public synchronized void append(OutboundEvent event) throws SQLException {
+        try (var statement = connection.prepareStatement(
+                "INSERT OR IGNORE INTO pending_events(event_id, payload, byte_size) VALUES (?, ?, ?)")) {
+            statement.setString(1, event.id());
+            statement.setString(2, event.payload());
+            statement.setInt(3, event.byteSize());
+            statement.executeUpdate();
+        }
+    }
+
+    public synchronized List<StoredEvent> peekBatch(int maxEvents, int maxBytes) throws SQLException {
+        if (maxEvents < 1 || maxBytes < 1) {
+            throw new IllegalArgumentException("Batch limits must be positive");
+        }
+        List<StoredEvent> events = new ArrayList<>();
+        int bytes = 0;
+        try (var statement = connection.prepareStatement("""
+                SELECT sequence, event_id, payload, byte_size
+                FROM pending_events
+                ORDER BY sequence
+                LIMIT ?
+                """)) {
+            statement.setInt(1, maxEvents);
+            try (var rows = statement.executeQuery()) {
+                while (rows.next()) {
+                    int eventBytes = rows.getInt("byte_size");
+                    if (bytes + eventBytes > maxBytes) {
+                        if (events.isEmpty()) {
+                            throw new SQLException(
+                                    "Outbox event " + rows.getString("event_id")
+                                            + " is larger than the configured batch byte limit");
+                        }
+                        break;
+                    }
+                    events.add(new StoredEvent(
+                            rows.getLong("sequence"),
+                            rows.getString("event_id"),
+                            rows.getString("payload"),
+                            eventBytes));
+                    bytes += eventBytes;
+                }
+            }
+        }
+        return List.copyOf(events);
+    }
+
+    public synchronized void acknowledgeThrough(long sequence) throws SQLException {
+        try (var statement = connection.prepareStatement(
+                "DELETE FROM pending_events WHERE sequence <= ?")) {
+            statement.setLong(1, sequence);
+            statement.executeUpdate();
+        }
+    }
+
+    public synchronized long pendingCount() throws SQLException {
+        try (var statement = connection.createStatement();
+                var rows = statement.executeQuery("SELECT COUNT(*) FROM pending_events")) {
+            return rows.getLong(1);
+        }
+    }
+
+    public synchronized long pendingBytes() throws SQLException {
+        try (var statement = connection.createStatement();
+                var rows = statement.executeQuery(
+                        "SELECT COALESCE(SUM(byte_size), 0) FROM pending_events")) {
+            return rows.getLong(1);
+        }
+    }
+
+    @Override
+    public synchronized void close() throws SQLException {
+        connection.close();
+    }
+}

src/main/java/games/dmg/spigotevents/outbox/StoredEvent.java

@@ -0,0 +1,4 @@
+package games.dmg.spigotevents.outbox;
+
+public record StoredEvent(long sequence, String eventId, String payload, int byteSize) {
+}

src/main/java/games/dmg/spigotevents/stats/MapChunker.java

@@ -0,0 +1,55 @@
+package games.dmg.spigotevents.stats;
+
+import com.google.gson.Gson;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+public final class MapChunker {
+    private static final Gson GSON = new Gson();
+
+    private MapChunker() {
+    }
+
+    public static <T> List<Map<String, T>> chunk(Map<String, T> values, int maxJsonBytes) {
+        if (maxJsonBytes < 2) {
+            throw new IllegalArgumentException("Maximum JSON bytes must be at least 2");
+        }
+        List<Map<String, T>> chunks = new ArrayList<>();
+        Map<String, T> current = new LinkedHashMap<>();
+        int currentBytes = 2; // Opening and closing braces.
+
+        for (var entry : values.entrySet()) {
+            Map<String, T> singleton = new LinkedHashMap<>();
+            singleton.put(entry.getKey(), entry.getValue());
+            int entryBytes = jsonBytes(singleton) - 2;
+            int separatorBytes = current.isEmpty() ? 0 : 1;
+            if (currentBytes + separatorBytes + entryBytes > maxJsonBytes) {
+                if (current.isEmpty()) {
+                    throw new IllegalArgumentException("One map entry exceeds the chunk byte limit");
+                }
+                chunks.add(immutableCopy(current));
+                current = new LinkedHashMap<>();
+                currentBytes = 2;
+                separatorBytes = 0;
+            }
+            current.put(entry.getKey(), entry.getValue());
+            currentBytes += separatorBytes + entryBytes;
+        }
+        if (!current.isEmpty()) {
+            chunks.add(immutableCopy(current));
+        }
+        return List.copyOf(chunks);
+    }
+
+    private static <T> Map<String, T> immutableCopy(Map<String, T> values) {
+        return Collections.unmodifiableMap(new LinkedHashMap<>(values));
+    }
+
+    private static int jsonBytes(Map<String, ?> values) {
+        return GSON.toJson(values).getBytes(StandardCharsets.UTF_8).length;
+    }
+}

src/main/java/games/dmg/spigotevents/stats/PlayerSnapshotCollector.java

@@ -0,0 +1,210 @@
+package games.dmg.spigotevents.stats;
+
+import games.dmg.spigotevents.delivery.EventPipeline;
+import games.dmg.spigotevents.event.CloudEventFactory;
+import games.dmg.spigotevents.event.PlayerIdentity;
+import java.time.Instant;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import org.bukkit.Location;
+import org.bukkit.Material;
+import org.bukkit.Statistic;
+import org.bukkit.entity.EntityType;
+import org.bukkit.entity.Player;
+
+public final class PlayerSnapshotCollector {
+    private static final Statistic[] STATISTICS = Statistic.values();
+    private static final Material[] MATERIALS = Material.values();
+    private static final EntityType[] ENTITY_TYPES = EntityType.values();
+
+    private final CloudEventFactory eventFactory;
+    private final EventPipeline pipeline;
+    private final int statisticChunkBytes;
+
+    public PlayerSnapshotCollector(
+            CloudEventFactory eventFactory,
+            EventPipeline pipeline,
+            int statisticChunkBytes) {
+        this.eventFactory = eventFactory;
+        this.pipeline = pipeline;
+        this.statisticChunkBytes = statisticChunkBytes;
+    }
+
+    public void captureLocation(Player player) {
+        Location location = player.getLocation();
+        Map<String, Object> data = new LinkedHashMap<>();
+        data.put("world_id", location.getWorld().getUID().toString());
+        data.put("world_name", location.getWorld().getName());
+        data.put("x", location.getX());
+        data.put("y", location.getY());
+        data.put("z", location.getZ());
+        data.put("yaw", location.getYaw());
+        data.put("pitch", location.getPitch());
+        data.put("game_mode", player.getGameMode().name().toLowerCase());
+        var biome = location.getBlock().getBiome();
+        var biomeKey = biome.getKeyOrNull();
+        data.put("biome", biomeKey == null ? biome.toString() : biomeKey.toString());
+        publish("player.location", identity(player), data);
+    }
+
+    public StatisticsCapture beginStatistics(Player player) {
+        return new StatisticsCapture(identity(player));
+    }
+
+    private void finishStatistics(StatisticsCapture capture) {
+        String completedAt = Instant.now().toString();
+        publishStatisticCategory(capture, completedAt, "untyped", capture.untyped);
+        publishStatisticCategory(capture, completedAt, "block", capture.blocks);
+        publishStatisticCategory(capture, completedAt, "item", capture.items);
+        publishStatisticCategory(capture, completedAt, "entity", capture.entities);
+    }
+
+    private void publishStatisticCategory(
+            StatisticsCapture capture,
+            String completedAt,
+            String category,
+            Map<String, Integer> statistics) {
+        List<Map<String, Integer>> chunks = MapChunker.chunk(statistics, statisticChunkBytes);
+        if (chunks.isEmpty()) {
+            chunks = List.of(Map.of());
+        }
+        for (int index = 0; index < chunks.size(); index++) {
+            Map<String, Object> data = new LinkedHashMap<>();
+            data.put("snapshot_id", capture.snapshotId.toString());
+            data.put("capture_started_at", capture.startedAt);
+            data.put("capture_completed_at", completedAt);
+            data.put("category", category);
+            data.put("part", index + 1);
+            data.put("parts", chunks.size());
+            data.put("statistics", chunks.get(index));
+            publish("player.statistics", capture.player, data);
+        }
+    }
+
+    private void publish(String type, PlayerIdentity player, Map<String, ?> data) {
+        pipeline.publish(eventFactory.create(type, player, data));
+    }
+
+    private static PlayerIdentity identity(Player player) {
+        return new PlayerIdentity(player.getUniqueId(), player.getName());
+    }
+
+    public final class StatisticsCapture {
+        private final PlayerIdentity player;
+        private final UUID snapshotId = UUID.randomUUID();
+        private final String startedAt = Instant.now().toString();
+        private final Map<String, Integer> untyped = new LinkedHashMap<>();
+        private final Map<String, Integer> blocks = new LinkedHashMap<>();
+        private final Map<String, Integer> items = new LinkedHashMap<>();
+        private final Map<String, Integer> entities = new LinkedHashMap<>();
+        private int statisticIndex;
+        private int subtypeIndex;
+        private boolean complete;
+
+        private StatisticsCapture(PlayerIdentity player) {
+            this.player = player;
+        }
+
+        public UUID playerId() {
+            return player.id();
+        }
+
+        public boolean process(Player currentPlayer, int operationBudget) {
+            if (complete) {
+                return true;
+            }
+            if (!currentPlayer.getUniqueId().equals(player.id())) {
+                throw new IllegalArgumentException("Statistics player does not match this capture");
+            }
+
+            int operations = 0;
+            while (operations < operationBudget && statisticIndex < STATISTICS.length) {
+                Statistic statistic = STATISTICS[statisticIndex];
+                switch (statistic.getType()) {
+                    case UNTYPED -> {
+                        readUntyped(currentPlayer, statistic, untyped);
+                        operations++;
+                        advanceStatistic();
+                    }
+                    case BLOCK -> operations += processMaterials(
+                            currentPlayer, statistic, blocks, true, operationBudget - operations);
+                    case ITEM -> operations += processMaterials(
+                            currentPlayer, statistic, items, false, operationBudget - operations);
+                    case ENTITY -> operations += processEntities(
+                            currentPlayer, statistic, entities, operationBudget - operations);
+                }
+            }
+
+            if (statisticIndex >= STATISTICS.length) {
+                complete = true;
+                finishStatistics(this);
+            }
+            return complete;
+        }
+
+        private int processMaterials(
+                Player currentPlayer,
+                Statistic statistic,
+                Map<String, Integer> target,
+                boolean blocksOnly,
+                int budget) {
+            int operations = 0;
+            while (operations < budget && subtypeIndex < MATERIALS.length) {
+                Material material = MATERIALS[subtypeIndex++];
+                operations++;
+                if ((blocksOnly && !material.isBlock()) || (!blocksOnly && !material.isItem())) {
+                    continue;
+                }
+                try {
+                    target.put(
+                            statistic.getKey() + "/" + material.getKeyOrThrow(),
+                            currentPlayer.getStatistic(statistic, material));
+                } catch (IllegalArgumentException ignored) {
+                    // Bukkit rejects combinations not represented by the game.
+                }
+            }
+            if (subtypeIndex >= MATERIALS.length) {
+                advanceStatistic();
+            }
+            return operations;
+        }
+
+        private int processEntities(
+                Player currentPlayer,
+                Statistic statistic,
+                Map<String, Integer> target,
+                int budget) {
+            int operations = 0;
+            while (operations < budget && subtypeIndex < ENTITY_TYPES.length) {
+                EntityType entityType = ENTITY_TYPES[subtypeIndex++];
+                operations++;
+                try {
+                    target.put(
+                            statistic.getKey() + "/" + entityType.getKeyOrThrow(),
+                            currentPlayer.getStatistic(statistic, entityType));
+                } catch (IllegalArgumentException ignored) {
+                    // Bukkit rejects unsupported entity types.
+                }
+            }
+            if (subtypeIndex >= ENTITY_TYPES.length) {
+                advanceStatistic();
+            }
+            return operations;
+        }
+
+        private void advanceStatistic() {
+            statisticIndex++;
+            subtypeIndex = 0;
+        }
+    }
+
+    private static void readUntyped(Player player, Statistic statistic, Map<String, Integer> target) {
+        try {
+            target.put(statistic.getKey().toString(), player.getStatistic(statistic));
+        } catch (IllegalArgumentException ignored) {
+            // Bukkit rejects statistics not represented by this game version.
+        }
+    }
+}

src/main/java/games/dmg/spigotevents/stats/SnapshotScheduler.java

@@ -0,0 +1,90 @@
+package games.dmg.spigotevents.stats;
+
+import java.util.ArrayDeque;
+import java.util.HashSet;
+import java.util.Queue;
+import java.util.Set;
+import java.util.UUID;
+import org.bukkit.Bukkit;
+import org.bukkit.entity.Player;
+import org.bukkit.plugin.Plugin;
+import org.bukkit.scheduler.BukkitTask;
+
+public final class SnapshotScheduler implements AutoCloseable {
+    private final Plugin plugin;
+    private final PlayerSnapshotCollector collector;
+    private final long intervalTicks;
+    private final int statisticOperationsPerTick;
+    private final Queue<UUID> statisticQueue = new ArrayDeque<>();
+    private final Set<UUID> queuedPlayers = new HashSet<>();
+    private PlayerSnapshotCollector.StatisticsCapture activeCapture;
+    private BukkitTask snapshotTask;
+    private BukkitTask statisticWorkerTask;
+
+    public SnapshotScheduler(
+            Plugin plugin,
+            PlayerSnapshotCollector collector,
+            long intervalSeconds,
+            int statisticOperationsPerTick) {
+        this.plugin = plugin;
+        this.collector = collector;
+        this.intervalTicks = Math.multiplyExact(intervalSeconds, 20L);
+        this.statisticOperationsPerTick = statisticOperationsPerTick;
+    }
+
+    public void start() {
+        snapshotTask = Bukkit.getScheduler().runTaskTimer(
+                plugin, this::beginSnapshot, intervalTicks, intervalTicks);
+        statisticWorkerTask = Bukkit.getScheduler().runTaskTimer(
+                plugin, this::continueStatistics, 1L, 1L);
+    }
+
+    private void beginSnapshot() {
+        for (Player player : Bukkit.getOnlinePlayers()) {
+            collector.captureLocation(player);
+            if (queuedPlayers.add(player.getUniqueId())) {
+                statisticQueue.add(player.getUniqueId());
+            }
+        }
+    }
+
+    private void continueStatistics() {
+        if (activeCapture == null) {
+            UUID nextPlayer = statisticQueue.poll();
+            if (nextPlayer == null) {
+                return;
+            }
+            Player player = Bukkit.getPlayer(nextPlayer);
+            if (player == null || !player.isOnline()) {
+                queuedPlayers.remove(nextPlayer);
+                return;
+            }
+            activeCapture = collector.beginStatistics(player);
+        }
+
+        UUID playerId = activeCapture.playerId();
+        Player player = Bukkit.getPlayer(playerId);
+        if (player == null || !player.isOnline()) {
+            queuedPlayers.remove(playerId);
+            activeCapture = null;
+            return;
+        }
+        if (activeCapture.process(player, statisticOperationsPerTick)) {
+            queuedPlayers.remove(playerId);
+            activeCapture = null;
+        }
+    }
+
+    @Override
+    public void close() {
+        if (snapshotTask != null) {
+            snapshotTask.cancel();
+        }
+        if (statisticWorkerTask != null) {
+            statisticWorkerTask.cancel();
+        }
+        activeCapture = null;
+        statisticQueue.clear();
+        queuedPlayers.clear();
+    }
+}

src/main/resources/config.yml

@@ -0,0 +1,28 @@
+# A human-readable stable name for this Minecraft server.
+server-name: "minecraft-server"
+
+# Generated once and persisted here when left empty. Do not copy the generated
+# value to another server; it is the stable identity used in CloudEvent source.
+server-id: ""
+
+ingest-url: "https://events.dmg.games/events"
+send-interval-seconds: 300
+snapshot-interval-seconds: 60
+http-timeout-seconds: 15
+
+batch:
+  max-events: 1000
+  # Leaves headroom below the ingest server's default 1 MiB request limit.
+  max-bytes: 900000
+
+# A send is also requested when either threshold is reached. This decision and
+# all HTTP/SQLite work happen on the outbox worker, never the server thread.
+dispatch-threshold:
+  events: 1000
+  bytes: 900000
+
+statistics:
+  # Statistic maps are partitioned before becoming CloudEvents.
+  chunk-json-bytes: 200000
+  # Maximum statistic/material/entity combinations read on the main thread per tick.
+  operations-per-tick: 1000

src/main/resources/plugin.yml

@@ -0,0 +1,6 @@
+name: SpigotEventProducer
+version: '${version}'
+main: games.dmg.spigotevents.SpigotEventProducerPlugin
+api-version: '26.2'
+author: dmg.games
+description: Writes Minecraft CloudEvents to a durable outbox and delivers them to game-ingest-server.