commit 12c03139473d49ed74dbb5d7431d5fd584a06265 Author: Dylan Garvis Date: Sat Aug 1 11:49:07 2026 -0400 feat: add Spigot event producer diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml new file mode 100644 index 0000000..8df692d --- /dev/null +++ b/.gitea/workflows/ci.yml @@ -0,0 +1,46 @@ +name: CI + +on: + push: + pull_request: + +jobs: + build: + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@v4 + with: + fetch-depth: 0 + + - name: Set up Java + uses: actions/setup-java@v4 + with: + distribution: temurin + java-version: 17 + cache: gradle + + - name: Validate conventional commits + if: github.event_name == 'pull_request' + run: | + npx -y \ + -p @commitlint/cli \ + -p @commitlint/config-conventional \ + commitlint --from "${{ github.event.pull_request.base.sha }}" --to "${{ github.sha }}" \ + --extends @commitlint/config-conventional + + - name: Build and test + run: ./gradlew clean check jar + + - name: Name development artifact + run: | + short_sha=$(printf '%s' "$GITHUB_SHA" | cut -c1-7) + cp build/libs/spigot-event-producer-0.1.0-SNAPSHOT.jar \ + "build/libs/spigot-event-producer-dev-${short_sha}.jar" + + - name: Upload development artifact + uses: actions/upload-artifact@v3 + with: + name: spigot-event-producer-${{ github.sha }} + path: build/libs/spigot-event-producer-dev-*.jar + if-no-files-found: error diff --git a/.gitea/workflows/release.yml b/.gitea/workflows/release.yml new file mode 100644 index 0000000..93493dd --- /dev/null +++ b/.gitea/workflows/release.yml @@ -0,0 +1,141 @@ +name: Release + +on: + push: + branches: + - main + +jobs: + release: + runs-on: ubuntu-latest + permissions: + contents: write + steps: + - name: Checkout + uses: actions/checkout@v4 + with: + fetch-depth: 0 + token: ${{ secrets.GITEA_TOKEN }} + + - name: Set up Java + uses: actions/setup-java@v4 + with: + distribution: temurin + java-version: 17 + cache: gradle + + - name: Set up Node.js + uses: actions/setup-node@v4 + with: + node-version: 20 + + - name: Build and test before release + run: ./gradlew clean check + + - name: Capture previous tag + id: previous_tag + run: | + if git describe --tags --abbrev=0 >/dev/null 2>&1; then + echo "value=$(git describe --tags --abbrev=0)" >> "$GITHUB_OUTPUT" + else + echo "value=" >> "$GITHUB_OUTPUT" + fi + + - name: Run semantic-release + env: + GITHUB_TOKEN: ${{ secrets.GITEA_TOKEN }} + run: | + npx -y \ + -p semantic-release \ + -p @semantic-release/commit-analyzer \ + -p @semantic-release/release-notes-generator \ + semantic-release \ + --branches main \ + --plugins @semantic-release/commit-analyzer,@semantic-release/release-notes-generator + + - name: Capture current tag + id: current_tag + run: | + if git describe --tags --exact-match HEAD >/dev/null 2>&1; then + echo "value=$(git describe --tags --exact-match HEAD)" >> "$GITHUB_OUTPUT" + else + echo "value=" >> "$GITHUB_OUTPUT" + fi + + - name: Resolve release + id: release + env: + PREVIOUS_TAG: ${{ steps.previous_tag.outputs.value }} + CURRENT_TAG: ${{ steps.current_tag.outputs.value }} + run: | + if [ -n "$CURRENT_TAG" ] && [ "$CURRENT_TAG" != "$PREVIOUS_TAG" ]; then + echo "created=true" >> "$GITHUB_OUTPUT" + echo "tag=$CURRENT_TAG" >> "$GITHUB_OUTPUT" + echo "version=${CURRENT_TAG#v}" >> "$GITHUB_OUTPUT" + else + echo "created=false" >> "$GITHUB_OUTPUT" + fi + + - name: Build versioned plugin + if: steps.release.outputs.created == 'true' + env: + VERSION: ${{ steps.release.outputs.version }} + run: ./gradlew clean jar -PreleaseVersion="$VERSION" + + - name: Upload release workflow artifact + if: steps.release.outputs.created == 'true' + uses: actions/upload-artifact@v3 + with: + name: spigot-event-producer-${{ steps.release.outputs.version }} + path: build/libs/spigot-event-producer-${{ steps.release.outputs.version }}.jar + if-no-files-found: error + + - name: Create Gitea release and upload plugin + if: steps.release.outputs.created == 'true' + env: + GITEA_TOKEN: ${{ secrets.GITEA_TOKEN }} + GITEA_SERVER_URL: ${{ github.server_url }} + REPOSITORY: ${{ github.repository }} + TAG: ${{ steps.release.outputs.tag }} + VERSION: ${{ steps.release.outputs.version }} + PREVIOUS_TAG: ${{ steps.previous_tag.outputs.value }} + run: | + api_url="${GITEA_SERVER_URL}/api/v1" + jar="build/libs/spigot-event-producer-${VERSION}.jar" + export RELEASE_BODY + if [ -n "$PREVIOUS_TAG" ]; then + RELEASE_BODY=$(git log --pretty='format:- %s (%h)' "${PREVIOUS_TAG}..HEAD") + else + RELEASE_BODY=$(git log --pretty='format:- %s (%h)' HEAD) + fi + payload=$(node -e ' + const payload = { + tag_name: process.env.TAG, + name: process.env.TAG, + body: process.env.RELEASE_BODY, + draft: false, + prerelease: false + }; + process.stdout.write(JSON.stringify(payload)); + ') + response=$(curl --fail-with-body --silent --show-error \ + -X POST \ + -H "Authorization: token ${GITEA_TOKEN}" \ + -H "Content-Type: application/json" \ + "${api_url}/repos/${REPOSITORY}/releases" \ + --data "$payload") + release_id=$(printf '%s' "$response" | node -e ' + let input = ""; + process.stdin.on("data", chunk => input += chunk); + process.stdin.on("end", () => { + const response = JSON.parse(input); + if (!response.id) process.exit(1); + process.stdout.write(String(response.id)); + }); + ') + curl --fail-with-body --silent --show-error \ + -X POST \ + -H "Authorization: token ${GITEA_TOKEN}" \ + -H "Content-Type: application/octet-stream" \ + --data-binary "@${jar}" \ + "${api_url}/repos/${REPOSITORY}/releases/${release_id}/assets?name=spigot-event-producer-${VERSION}.jar" diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..adb24a4 --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +.gradle/ +build/ +*.iml +.idea/ +.vscode/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..835765f --- /dev/null +++ b/README.md @@ -0,0 +1,129 @@ +# Spigot Event Producer + +A Spigot 26.2 plugin that records Minecraft activity as CloudEvents 1.0, persists it in a local SQLite outbox, and sends newline-delimited batches to `game-ingest-server`. + +## Requirements + +- Spigot 26.2 +- Java 17 or newer +- Network access to the ingest endpoint + +## Build and install + +```bash +./gradlew clean test jar +cp build/libs/spigot-event-producer-0.1.0-SNAPSHOT.jar /path/to/server/plugins/ +``` + +Start the Minecraft server once to create `plugins/SpigotEventProducer/config.yml`. Set a meaningful, stable `server-name`. When `server-id` is empty, the plugin generates and persists a UUID. Do not reuse that generated UUID on another server. + +The default ingest endpoint is: + +```text +https://events.dmg.games/events +``` + +## Events + +Every event is a structured CloudEvent 1.0 JSON object with these common attributes: + +- `source`: `urn:minecraft-server:` +- `serverid` and `servername`: CloudEvent extensions identifying the server +- `gameversion`: the Bukkit game version +- `playerid`: player UUID for player events +- `data.player_id` and `data.player_name`: stable UUID and current player name + +Produced event types: + +| Type | Trigger | +| --- | --- | +| `games.dmg.minecraft.server.started` | Plugin/server startup | +| `games.dmg.minecraft.server.stopped` | Plugin/server shutdown | +| `games.dmg.minecraft.player.joined` | Player joins | +| `games.dmg.minecraft.player.quit` | Player quits | +| `games.dmg.minecraft.chat` | Non-cancelled public player chat | +| `games.dmg.minecraft.private_message` | A recognized private-message command | +| `games.dmg.minecraft.player.location` | Location snapshot | +| `games.dmg.minecraft.player.statistics` | Partitioned statistic snapshot | + +### Chat and private messages + +Public chat events contain the complete message text. + +Private-message events recognize `/msg`, `/message`, `/tell`, `/w`, `/whisper`, `/pm`, `/reply`, and `/r`, including namespaced variants such as `/minecraft:tell`. Explicit commands contain the destination token. Reply commands set `reply: true` and omit the recipient so downstream processing can infer it from earlier ordered events. The event also reports whether the command was cancelled. + +Spigot has no universal private-message event. Commands implemented with unrelated aliases or entirely custom plugin logic will not be recognized until their syntax is added. A recognized command event reports observation of the command, not guaranteed delivery to its recipient. + +### Location snapshots + +Each online player is sampled every 60 seconds by default. Events include world UUID/name, coordinates, yaw, pitch, game mode, and biome. + +### Statistic snapshots + +Every Bukkit statistic is read, including untyped, block, item, and entity statistics. Zero values are retained. Invalid statistic/material/entity combinations rejected by Bukkit are skipped. + +Large snapshots are divided by category and size. All parts share a `snapshot_id`, capture start/completion timestamps, and include `part` and `parts` fields. Statistic reads are spread across ticks using a configurable operation budget; a player remains queued until all combinations have been read. Empty categories still emit an explicit part. + +## Delivery and durability + +The outbox is stored at: + +```text +plugins/SpigotEventProducer/event-outbox.sqlite3 +``` + +SQLite writes and queue-size decisions run on a dedicated persistence worker. NDJSON construction and HTTP requests use a separate delivery worker, so a slow ingest request cannot prevent newly observed events from reaching SQLite. Bukkit state is only read on Bukkit-managed threads. + +The worker sends every five minutes by default. It may send earlier when the configured event or byte threshold is reached. Requests remain below the ingest service's default 1 MiB and 1,000-event limits. Events are removed only after HTTP `202 Accepted`; failures remain in SQLite and are retried. + +Delivery is at least once. If the ingest server accepts a request but its response is lost, the plugin retries the same stable CloudEvent IDs. Consumers should deduplicate by CloudEvent `id` when necessary. + +A shutdown event is persisted before the plugin closes its outbox. If it is not uploaded during shutdown, it is delivered after the next startup. + +## Configuration + +```yaml +server-name: "minecraft-server" +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 + max-bytes: 900000 + +dispatch-threshold: + events: 1000 + bytes: 900000 + +statistics: + chunk-json-bytes: 200000 + operations-per-tick: 1000 +``` + +All intervals and limits must be positive. Keep `batch.max-bytes` below the ingest server's configured body limit and `batch.max-events` at or below its maximum events per request. + +## Releases + +Gitea Actions runs the Gradle checks for every push and pull request and stores a development JAR as a workflow artifact. Pushes to `main` also run semantic-release using conventional commits: + +- `fix:` creates a patch release; +- `feat:` creates a minor release; +- a breaking-change footer or `!` creates a major release. + +A successful release creates a `vX.Y.Z` tag and attaches `spigot-event-producer-X.Y.Z.jar` permanently to the corresponding Gitea Release. The release workflow requires a repository secret named `GITEA_TOKEN` with repository contents write permission. + +For a local versioned build: + +```bash +./gradlew clean check jar -PreleaseVersion=1.2.3 +``` + +## Privacy and operations + +This plugin deliberately records complete public and private message text, player identifiers, precise locations, and activity statistics. Server operators should disclose that collection, restrict access to the plugin data directory and ingest database, use HTTPS, and configure retention appropriate to their jurisdiction. + +The outbox has no automatic retention limit because dropping unsent events would violate durability. Monitor disk usage if the ingest service is unavailable for an extended period. diff --git a/build.gradle.kts b/build.gradle.kts new file mode 100644 index 0000000..61abd41 --- /dev/null +++ b/build.gradle.kts @@ -0,0 +1,54 @@ +plugins { + java +} + +group = "games.dmg" +version = providers.gradleProperty("releaseVersion") + .orElse("0.1.0-SNAPSHOT") + .get() + +repositories { + mavenCentral() + maven("https://hub.spigotmc.org/nexus/content/repositories/snapshots/") +} + +java { + toolchain { + languageVersion = JavaLanguageVersion.of(17) + } +} + +tasks.withType().configureEach { + options.compilerArgs.add("-Xlint:deprecation") +} + +dependencies { + compileOnly("org.spigotmc:spigot-api:26.2-R0.1-SNAPSHOT") + implementation("org.xerial:sqlite-jdbc:3.50.3.0") + + testImplementation("org.spigotmc:spigot-api:26.2-R0.1-SNAPSHOT") + testImplementation(platform("org.junit:junit-bom:5.13.4")) + testImplementation("org.junit.jupiter:junit-jupiter") + testRuntimeOnly("org.junit.platform:junit-platform-launcher") +} + +tasks.test { + useJUnitPlatform() +} + +val pluginVersion = version + +tasks.processResources { + filesMatching("plugin.yml") { + expand("version" to pluginVersion) + } +} + +tasks.jar { + duplicatesStrategy = DuplicatesStrategy.EXCLUDE + manifest { + attributes["Multi-Release"] = "true" + } + from(configurations.runtimeClasspath.get().map { if (it.isDirectory) it else zipTree(it) }) + exclude("META-INF/*.SF", "META-INF/*.DSA", "META-INF/*.RSA") +} diff --git a/gradle/wrapper/gradle-wrapper.jar b/gradle/wrapper/gradle-wrapper.jar new file mode 100644 index 0000000..b1b8ef5 Binary files /dev/null and b/gradle/wrapper/gradle-wrapper.jar differ diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties new file mode 100644 index 0000000..a351597 --- /dev/null +++ b/gradle/wrapper/gradle-wrapper.properties @@ -0,0 +1,7 @@ +distributionBase=GRADLE_USER_HOME +distributionPath=wrapper/dists +distributionUrl=https\://services.gradle.org/distributions/gradle-9.6.1-bin.zip +networkTimeout=10000 +validateDistributionUrl=true +zipStoreBase=GRADLE_USER_HOME +zipStorePath=wrapper/dists diff --git a/gradlew b/gradlew new file mode 100755 index 0000000..203529c --- /dev/null +++ b/gradlew @@ -0,0 +1,248 @@ +#!/bin/sh + +# +# Copyright © 2015 the original authors. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# SPDX-License-Identifier: Apache-2.0 +# + +############################################################################## +# +# gradlew start up script for POSIX generated by Gradle. +# +# Important for running: +# +# (1) You need a POSIX-compliant shell to run this script. If your /bin/sh is +# noncompliant, but you have some other compliant shell such as ksh or +# bash, then to run this script, type that shell name before the whole +# command line, like: +# +# ksh gradlew +# +# Busybox and similar reduced shells will NOT work, because this script +# requires all of these POSIX shell features: +# * functions; +# * expansions «$var», «${var}», «${var:-default}», «${var+SET}», +# «${var#prefix}», «${var%suffix}», and «$( cmd )»; +# * compound commands having a testable exit status, especially «case»; +# * various built-in commands including «command», «set», and «ulimit». +# +# Important for patching: +# +# (2) This script targets any POSIX shell, so it avoids extensions provided +# by Bash, Ksh, etc; in particular arrays are avoided. +# +# The "traditional" practice of packing multiple parameters into a +# space-separated string is a well documented source of bugs and security +# problems, so this is (mostly) avoided, by progressively accumulating +# options in "$@", and eventually passing that to Java. +# +# Where the inherited environment variables (DEFAULT_JVM_OPTS, JAVA_OPTS, +# and GRADLE_OPTS) rely on word-splitting, this is performed explicitly; +# see the in-line comments for details. +# +# There are tweaks for specific operating systems such as AIX, CygWin, +# Darwin, MinGW, and NonStop. +# +# (3) This script is generated from the Groovy template +# https://github.com/gradle/gradle/blob/3d91ce3b8caaf77ad09f381f43615b715b53f72c/platforms/jvm/plugins-application/src/main/resources/org/gradle/api/internal/plugins/unixStartScript.txt +# within the Gradle project. +# +# You can find Gradle at https://github.com/gradle/gradle/. +# +############################################################################## + +# Attempt to set APP_HOME + +# Resolve links: $0 may be a link +app_path=$0 + +# Need this for daisy-chained symlinks. +while + APP_HOME=${app_path%"${app_path##*/}"} # leaves a trailing /; empty if no leading path + [ -h "$app_path" ] +do + ls=$( ls -ld "$app_path" ) + link=${ls#*' -> '} + case $link in #( + /*) app_path=$link ;; #( + *) app_path=$APP_HOME$link ;; + esac +done + +# This is normally unused +# shellcheck disable=SC2034 +APP_BASE_NAME=${0##*/} +# Discard cd standard output in case $CDPATH is set (https://github.com/gradle/gradle/issues/25036) +APP_HOME=$( cd -P "${APP_HOME:-./}" > /dev/null && printf '%s\n' "$PWD" ) || exit + +# Use the maximum available, or set MAX_FD != -1 to use that value. +MAX_FD=maximum + +warn () { + echo "$*" +} >&2 + +die () { + echo + echo "$*" + echo + exit 1 +} >&2 + +# OS specific support (must be 'true' or 'false'). +cygwin=false +msys=false +darwin=false +nonstop=false +case "$( uname )" in #( + CYGWIN* ) cygwin=true ;; #( + Darwin* ) darwin=true ;; #( + MSYS* | MINGW* ) msys=true ;; #( + NONSTOP* ) nonstop=true ;; +esac + + + +# Determine the Java command to use to start the JVM. +if [ -n "$JAVA_HOME" ] ; then + if [ -x "$JAVA_HOME/jre/sh/java" ] ; then + # IBM's JDK on AIX uses strange locations for the executables + JAVACMD=$JAVA_HOME/jre/sh/java + else + JAVACMD=$JAVA_HOME/bin/java + fi + if [ ! -x "$JAVACMD" ] ; then + die "ERROR: JAVA_HOME is set to an invalid directory: $JAVA_HOME + +Please set the JAVA_HOME variable in your environment to match the +location of your Java installation." + fi +else + JAVACMD=java + if ! command -v java >/dev/null 2>&1 + then + die "ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH. + +Please set the JAVA_HOME variable in your environment to match the +location of your Java installation." + fi +fi + +# Increase the maximum file descriptors if we can. +if ! "$cygwin" && ! "$darwin" && ! "$nonstop" ; then + case $MAX_FD in #( + max*) + # In POSIX sh, ulimit -H is undefined. That's why the result is checked to see if it worked. + # shellcheck disable=SC2039,SC3045 + MAX_FD=$( ulimit -H -n ) || + warn "Could not query maximum file descriptor limit" + esac + case $MAX_FD in #( + '' | soft) :;; #( + *) + # In POSIX sh, ulimit -n is undefined. That's why the result is checked to see if it worked. + # shellcheck disable=SC2039,SC3045 + ulimit -n "$MAX_FD" || + warn "Could not set maximum file descriptor limit to $MAX_FD" + esac +fi + +# Collect all arguments for the java command, stacking in reverse order: +# * args from the command line +# * the main class name +# * -classpath +# * -D...appname settings +# * --module-path (only if needed) +# * DEFAULT_JVM_OPTS, JAVA_OPTS, and GRADLE_OPTS environment variables. + +# For Cygwin or MSYS, switch paths to Windows format before running java +if "$cygwin" || "$msys" ; then + APP_HOME=$( cygpath --path --mixed "$APP_HOME" ) + + JAVACMD=$( cygpath --unix "$JAVACMD" ) + + # Now convert the arguments - kludge to limit ourselves to /bin/sh + for arg do + if + case $arg in #( + -*) false ;; # don't mess with options #( + /?*) t=${arg#/} t=/${t%%/*} # looks like a POSIX filepath + [ -e "$t" ] ;; #( + *) false ;; + esac + then + arg=$( cygpath --path --ignore --mixed "$arg" ) + fi + # Roll the args list around exactly as many times as the number of + # args, so each arg winds up back in the position where it started, but + # possibly modified. + # + # NB: a `for` loop captures its iteration list before it begins, so + # changing the positional parameters here affects neither the number of + # iterations, nor the values presented in `arg`. + shift # remove old arg + set -- "$@" "$arg" # push replacement arg + done +fi + + +# Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script. +DEFAULT_JVM_OPTS='-Dfile.encoding=UTF-8 "-Xmx64m" "-Xms64m"' + +# Collect all arguments for the java command: +# * DEFAULT_JVM_OPTS, JAVA_OPTS, and optsEnvironmentVar are not allowed to contain shell fragments, +# and any embedded shellness will be escaped. +# * For example: A user cannot expect ${Hostname} to be expanded, as it is an environment variable and will be +# treated as '${Hostname}' itself on the command line. + +set -- \ + "-Dorg.gradle.appname=$APP_BASE_NAME" \ + -jar "$APP_HOME/gradle/wrapper/gradle-wrapper.jar" \ + "$@" + +# Stop when "xargs" is not available. +if ! command -v xargs >/dev/null 2>&1 +then + die "xargs is not available" +fi + +# Use "xargs" to parse quoted args. +# +# With -n1 it outputs one arg per line, with the quotes and backslashes removed. +# +# In Bash we could simply go: +# +# readarray ARGS < <( xargs -n1 <<<"$var" ) && +# set -- "${ARGS[@]}" "$@" +# +# but POSIX shell has neither arrays nor command substitution, so instead we +# post-process each arg (as a line of input to sed) to backslash-escape any +# character that might be a shell metacharacter, then use eval to reverse +# that process (while maintaining the separation between arguments), and wrap +# the whole thing up as a single "set" statement. +# +# This will of course break if any of these variables contains a newline or +# an unmatched quote. +# + +eval "set -- $( + printf '%s\n' "$DEFAULT_JVM_OPTS $JAVA_OPTS $GRADLE_OPTS" | + xargs -n1 | + sed ' s~[^-[:alnum:]+,./:=@_]~\\&~g; ' | + tr '\n' ' ' + )" '"$@"' + +exec "$JAVACMD" "$@" diff --git a/gradlew.bat b/gradlew.bat new file mode 100644 index 0000000..7e60b72 --- /dev/null +++ b/gradlew.bat @@ -0,0 +1,82 @@ +@rem +@rem Copyright 2015 the original author or authors. +@rem +@rem Licensed under the Apache License, Version 2.0 (the "License"); +@rem you may not use this file except in compliance with the License. +@rem You may obtain a copy of the License at +@rem +@rem https://www.apache.org/licenses/LICENSE-2.0 +@rem +@rem Unless required by applicable law or agreed to in writing, software +@rem distributed under the License is distributed on an "AS IS" BASIS, +@rem WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +@rem See the License for the specific language governing permissions and +@rem limitations under the License. +@rem +@rem SPDX-License-Identifier: Apache-2.0 +@rem + +@if "%DEBUG%"=="" @echo off +@rem ########################################################################## +@rem +@rem gradlew startup script for Windows +@rem +@rem ########################################################################## + +@rem Set local scope for the variables, and ensure extensions are enabled +setlocal EnableExtensions + +set DIRNAME=%~dp0 +if "%DIRNAME%"=="" set DIRNAME=. +@rem This is normally unused +set APP_BASE_NAME=%~n0 +set APP_HOME=%DIRNAME% + +@rem Resolve any "." and ".." in APP_HOME to make it shorter. +for %%i in ("%APP_HOME%") do set APP_HOME=%%~fi + +@rem Add default JVM options here. You can also use JAVA_OPTS and GRADLE_OPTS to pass JVM options to this script. +set DEFAULT_JVM_OPTS=-Dfile.encoding=UTF-8 "-Xmx64m" "-Xms64m" + +@rem Find java.exe +if defined JAVA_HOME goto findJavaFromJavaHome + +set JAVA_EXE=java.exe +%JAVA_EXE% -version >NUL 2>&1 +if %ERRORLEVEL% equ 0 goto execute + +echo. 1>&2 +echo ERROR: JAVA_HOME is not set and no 'java' command could be found in your PATH. 1>&2 +echo. 1>&2 +echo Please set the JAVA_HOME variable in your environment to match the 1>&2 +echo location of your Java installation. 1>&2 + +"%COMSPEC%" /c exit 1 + +:findJavaFromJavaHome +set JAVA_HOME=%JAVA_HOME:"=% +set JAVA_EXE=%JAVA_HOME%/bin/java.exe + +if exist "%JAVA_EXE%" goto execute + +echo. 1>&2 +echo ERROR: JAVA_HOME is set to an invalid directory: %JAVA_HOME% 1>&2 +echo. 1>&2 +echo Please set the JAVA_HOME variable in your environment to match the 1>&2 +echo location of your Java installation. 1>&2 + +"%COMSPEC%" /c exit 1 + +:execute +@rem Setup the command line + + + +@rem Execute gradlew +@rem endlocal doesn't take effect until after the line is parsed and variables are expanded +@rem which allows us to clear the local environment before executing the java command +endlocal & "%JAVA_EXE%" %DEFAULT_JVM_OPTS% %JAVA_OPTS% %GRADLE_OPTS% "-Dorg.gradle.appname=%APP_BASE_NAME%" -jar "%APP_HOME%\gradle\wrapper\gradle-wrapper.jar" %* & call :exitWithErrorLevel + +:exitWithErrorLevel +@rem Use "%COMSPEC%" /c exit to allow operators to work properly in scripts +"%COMSPEC%" /c exit %ERRORLEVEL% diff --git a/settings.gradle.kts b/settings.gradle.kts new file mode 100644 index 0000000..8bf41e0 --- /dev/null +++ b/settings.gradle.kts @@ -0,0 +1 @@ +rootProject.name = "spigot-event-producer" diff --git a/src/main/java/games/dmg/spigotevents/SpigotEventProducerPlugin.java b/src/main/java/games/dmg/spigotevents/SpigotEventProducerPlugin.java new file mode 100644 index 0000000..4aa3fe0 --- /dev/null +++ b/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 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) { + } +} diff --git a/src/main/java/games/dmg/spigotevents/chat/PrivateMessageParser.java b/src/main/java/games/dmg/spigotevents/chat/PrivateMessageParser.java new file mode 100644 index 0000000..5dbc576 --- /dev/null +++ b/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 EXPLICIT_COMMANDS = Set.of( + "msg", "message", "tell", "w", "whisper", "pm"); + private static final Set REPLY_COMMANDS = Set.of("reply", "r"); + + private PrivateMessageParser() { + } + + public static Optional 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) { + } +} diff --git a/src/main/java/games/dmg/spigotevents/delivery/BatchSender.java b/src/main/java/games/dmg/spigotevents/delivery/BatchSender.java new file mode 100644 index 0000000..5b52dc9 --- /dev/null +++ b/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 events) throws Exception; +} diff --git a/src/main/java/games/dmg/spigotevents/delivery/EventPipeline.java b/src/main/java/games/dmg/spigotevents/delivery/EventPipeline.java new file mode 100644 index 0000000..f3a300a --- /dev/null +++ b/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 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 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; + } +} diff --git a/src/main/java/games/dmg/spigotevents/delivery/IngestClient.java b/src/main/java/games/dmg/spigotevents/delivery/IngestClient.java new file mode 100644 index 0000000..352e84c --- /dev/null +++ b/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 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 response = client.send(request, HttpResponse.BodyHandlers.discarding()); + return response.statusCode() == 202; + } +} diff --git a/src/main/java/games/dmg/spigotevents/delivery/OutboxDispatcher.java b/src/main/java/games/dmg/spigotevents/delivery/OutboxDispatcher.java new file mode 100644 index 0000000..d1a59a6 --- /dev/null +++ b/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; + } +} diff --git a/src/main/java/games/dmg/spigotevents/event/CloudEventFactory.java b/src/main/java/games/dmg/spigotevents/event/CloudEventFactory.java new file mode 100644 index 0000000..81203cd --- /dev/null +++ b/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 eventIds; + + public CloudEventFactory(ServerIdentity server, String gameVersion) { + this(server, gameVersion, Clock.systemUTC(), UUID::randomUUID); + } + + CloudEventFactory( + ServerIdentity server, + String gameVersion, + Clock clock, + Supplier 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 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 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 eventData) { + return create(eventName, null, eventData); + } +} diff --git a/src/main/java/games/dmg/spigotevents/event/OutboundEvent.java b/src/main/java/games/dmg/spigotevents/event/OutboundEvent.java new file mode 100644 index 0000000..2c6d422 --- /dev/null +++ b/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; + } +} diff --git a/src/main/java/games/dmg/spigotevents/event/PlayerIdentity.java b/src/main/java/games/dmg/spigotevents/event/PlayerIdentity.java new file mode 100644 index 0000000..edde920 --- /dev/null +++ b/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"); + } +} diff --git a/src/main/java/games/dmg/spigotevents/event/ServerIdentity.java b/src/main/java/games/dmg/spigotevents/event/ServerIdentity.java new file mode 100644 index 0000000..8f4f93f --- /dev/null +++ b/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"); + } + } +} diff --git a/src/main/java/games/dmg/spigotevents/listener/GameEventListener.java b/src/main/java/games/dmg/spigotevents/listener/GameEventListener.java new file mode 100644 index 0000000..9d624df --- /dev/null +++ b/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 playerIdentities = + Collections.synchronizedMap(new IdentityHashMap<>()); + + public GameEventListener( + CloudEventFactory eventFactory, + EventPipeline pipeline, + Collection 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 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 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)); + } +} diff --git a/src/main/java/games/dmg/spigotevents/outbox/SqliteOutbox.java b/src/main/java/games/dmg/spigotevents/outbox/SqliteOutbox.java new file mode 100644 index 0000000..1765653 --- /dev/null +++ b/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 peekBatch(int maxEvents, int maxBytes) throws SQLException { + if (maxEvents < 1 || maxBytes < 1) { + throw new IllegalArgumentException("Batch limits must be positive"); + } + List 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(); + } +} diff --git a/src/main/java/games/dmg/spigotevents/outbox/StoredEvent.java b/src/main/java/games/dmg/spigotevents/outbox/StoredEvent.java new file mode 100644 index 0000000..9b07fb3 --- /dev/null +++ b/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) { +} diff --git a/src/main/java/games/dmg/spigotevents/stats/MapChunker.java b/src/main/java/games/dmg/spigotevents/stats/MapChunker.java new file mode 100644 index 0000000..17601f4 --- /dev/null +++ b/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 List> chunk(Map values, int maxJsonBytes) { + if (maxJsonBytes < 2) { + throw new IllegalArgumentException("Maximum JSON bytes must be at least 2"); + } + List> chunks = new ArrayList<>(); + Map current = new LinkedHashMap<>(); + int currentBytes = 2; // Opening and closing braces. + + for (var entry : values.entrySet()) { + Map 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 Map immutableCopy(Map values) { + return Collections.unmodifiableMap(new LinkedHashMap<>(values)); + } + + private static int jsonBytes(Map values) { + return GSON.toJson(values).getBytes(StandardCharsets.UTF_8).length; + } +} diff --git a/src/main/java/games/dmg/spigotevents/stats/PlayerSnapshotCollector.java b/src/main/java/games/dmg/spigotevents/stats/PlayerSnapshotCollector.java new file mode 100644 index 0000000..b0c2558 --- /dev/null +++ b/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 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 statistics) { + List> chunks = MapChunker.chunk(statistics, statisticChunkBytes); + if (chunks.isEmpty()) { + chunks = List.of(Map.of()); + } + for (int index = 0; index < chunks.size(); index++) { + Map 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 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 untyped = new LinkedHashMap<>(); + private final Map blocks = new LinkedHashMap<>(); + private final Map items = new LinkedHashMap<>(); + private final Map 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 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 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 target) { + try { + target.put(statistic.getKey().toString(), player.getStatistic(statistic)); + } catch (IllegalArgumentException ignored) { + // Bukkit rejects statistics not represented by this game version. + } + } +} diff --git a/src/main/java/games/dmg/spigotevents/stats/SnapshotScheduler.java b/src/main/java/games/dmg/spigotevents/stats/SnapshotScheduler.java new file mode 100644 index 0000000..a442a7b --- /dev/null +++ b/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 statisticQueue = new ArrayDeque<>(); + private final Set 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(); + } +} diff --git a/src/main/resources/config.yml b/src/main/resources/config.yml new file mode 100644 index 0000000..1b99702 --- /dev/null +++ b/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 diff --git a/src/main/resources/plugin.yml b/src/main/resources/plugin.yml new file mode 100644 index 0000000..6d2e56e --- /dev/null +++ b/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. diff --git a/src/test/java/games/dmg/spigotevents/chat/PrivateMessageParserTest.java b/src/test/java/games/dmg/spigotevents/chat/PrivateMessageParserTest.java new file mode 100644 index 0000000..0446929 --- /dev/null +++ b/src/test/java/games/dmg/spigotevents/chat/PrivateMessageParserTest.java @@ -0,0 +1,27 @@ +package games.dmg.spigotevents.chat; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.junit.jupiter.api.Test; + +class PrivateMessageParserTest { + @Test + void parsesAnExplicitPrivateMessageWithoutLosingItsText() { + var parsed = PrivateMessageParser.parse("/msg Alex hello from spawn"); + + assertTrue(parsed.isPresent()); + assertEquals("msg", parsed.orElseThrow().command()); + assertEquals("Alex", parsed.orElseThrow().recipient()); + assertEquals("hello from spawn", parsed.orElseThrow().message()); + } + + @Test + void representsReplyCommandsWithoutInventingARecipient() { + var parsed = PrivateMessageParser.parse("/reply same place as before"); + + assertTrue(parsed.isPresent()); + assertTrue(parsed.orElseThrow().recipient().isEmpty()); + assertEquals("same place as before", parsed.orElseThrow().message()); + } +} diff --git a/src/test/java/games/dmg/spigotevents/delivery/EventPipelineTest.java b/src/test/java/games/dmg/spigotevents/delivery/EventPipelineTest.java new file mode 100644 index 0000000..4415433 --- /dev/null +++ b/src/test/java/games/dmg/spigotevents/delivery/EventPipelineTest.java @@ -0,0 +1,51 @@ +package games.dmg.spigotevents.delivery; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import games.dmg.spigotevents.event.OutboundEvent; +import games.dmg.spigotevents.outbox.SqliteOutbox; +import java.nio.file.Path; +import java.time.Duration; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +class EventPipelineTest { + @TempDir + Path temporaryDirectory; + + @Test + void persistsNewEventsWhileAnHttpDeliveryIsBlocked() throws Exception { + CountDownLatch deliveryStarted = new CountDownLatch(1); + CountDownLatch releaseDelivery = new CountDownLatch(1); + try (var outbox = new SqliteOutbox(temporaryDirectory.resolve("outbox.db"))) { + BatchSender blockedSender = events -> { + deliveryStarted.countDown(); + releaseDelivery.await(5, TimeUnit.SECONDS); + return false; + }; + var dispatcher = new OutboxDispatcher(outbox, blockedSender, 1_000, 900_000); + var pipeline = new EventPipeline( + outbox, + dispatcher, + Duration.ofMinutes(5), + 1, + 900_000, + (message, error) -> { }); + try { + pipeline.publish(new OutboundEvent("one", "{\"id\":\"one\"}")); + assertTrue(deliveryStarted.await(2, TimeUnit.SECONDS)); + + pipeline.publishAndWait( + new OutboundEvent("two", "{\"id\":\"two\"}"), Duration.ofSeconds(2)); + + assertEquals(2, outbox.pendingCount()); + } finally { + releaseDelivery.countDown(); + pipeline.close(); + } + } + } +} diff --git a/src/test/java/games/dmg/spigotevents/delivery/OutboxDeliveryTest.java b/src/test/java/games/dmg/spigotevents/delivery/OutboxDeliveryTest.java new file mode 100644 index 0000000..313cd26 --- /dev/null +++ b/src/test/java/games/dmg/spigotevents/delivery/OutboxDeliveryTest.java @@ -0,0 +1,58 @@ +package games.dmg.spigotevents.delivery; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import com.sun.net.httpserver.HttpServer; +import games.dmg.spigotevents.event.OutboundEvent; +import games.dmg.spigotevents.outbox.SqliteOutbox; +import java.net.InetSocketAddress; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.nio.file.Path; +import java.time.Duration; +import java.util.ArrayDeque; +import java.util.Queue; +import java.util.concurrent.atomic.AtomicReference; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +class OutboxDeliveryTest { + @TempDir + Path temporaryDirectory; + + @Test + void sendsNdjsonAndOnlyAcknowledgesAnAcceptedBatch() throws Exception { + Queue responses = new ArrayDeque<>(); + responses.add(500); + responses.add(202); + AtomicReference requestBody = new AtomicReference<>(); + AtomicReference contentType = new AtomicReference<>(); + HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/events", exchange -> { + requestBody.set(new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8)); + contentType.set(exchange.getRequestHeaders().getFirst("Content-Type")); + int status = responses.remove(); + exchange.sendResponseHeaders(status, -1); + exchange.close(); + }); + server.start(); + + try (var outbox = new SqliteOutbox(temporaryDirectory.resolve("outbox.db"))) { + outbox.append(new OutboundEvent("one", "{\"id\":\"one\"}")); + var client = new IngestClient( + URI.create("http://127.0.0.1:" + server.getAddress().getPort() + "/events"), + Duration.ofSeconds(2)); + var dispatcher = new OutboxDispatcher(outbox, client, 1_000, 900_000); + + dispatcher.dispatchOnce(); + assertEquals(1, outbox.pendingCount()); + + dispatcher.dispatchOnce(); + assertEquals(0, outbox.pendingCount()); + assertEquals("application/x-ndjson", contentType.get()); + assertEquals("{\"id\":\"one\"}\n", requestBody.get()); + } finally { + server.stop(0); + } + } +} diff --git a/src/test/java/games/dmg/spigotevents/event/CloudEventFactoryTest.java b/src/test/java/games/dmg/spigotevents/event/CloudEventFactoryTest.java new file mode 100644 index 0000000..60d7d8b --- /dev/null +++ b/src/test/java/games/dmg/spigotevents/event/CloudEventFactoryTest.java @@ -0,0 +1,40 @@ +package games.dmg.spigotevents.event; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import com.google.gson.JsonParser; +import java.time.Clock; +import java.time.Instant; +import java.time.ZoneOffset; +import java.util.Map; +import java.util.UUID; +import org.junit.jupiter.api.Test; + +class CloudEventFactoryTest { + @Test + void createsAnIngestCompatiblePlayerCloudEvent() { + UUID serverId = UUID.fromString("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"); + UUID playerId = UUID.fromString("bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"); + var factory = new CloudEventFactory( + new ServerIdentity(serverId, "survival-one"), + "26.2", + Clock.fixed(Instant.parse("2026-08-01T10:15:30Z"), ZoneOffset.UTC), + () -> UUID.fromString("cccccccc-cccc-cccc-cccc-cccccccccccc")); + + OutboundEvent event = factory.create( + "chat", + new PlayerIdentity(playerId, "Alex"), + Map.of("message", "hello \"world\"")); + var json = JsonParser.parseString(event.payload()).getAsJsonObject(); + + assertEquals("1.0", json.get("specversion").getAsString()); + assertEquals("cccccccc-cccc-cccc-cccc-cccccccccccc", json.get("id").getAsString()); + assertEquals("urn:minecraft-server:aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", json.get("source").getAsString()); + assertEquals("games.dmg.minecraft.chat", json.get("type").getAsString()); + assertEquals("2026-08-01T10:15:30Z", json.get("time").getAsString()); + assertEquals("survival-one", json.get("servername").getAsString()); + assertEquals(playerId.toString(), json.get("playerid").getAsString()); + assertEquals("Alex", json.getAsJsonObject("data").get("player_name").getAsString()); + assertEquals("hello \"world\"", json.getAsJsonObject("data").get("message").getAsString()); + } +} diff --git a/src/test/java/games/dmg/spigotevents/outbox/SqliteOutboxTest.java b/src/test/java/games/dmg/spigotevents/outbox/SqliteOutboxTest.java new file mode 100644 index 0000000..1f60139 --- /dev/null +++ b/src/test/java/games/dmg/spigotevents/outbox/SqliteOutboxTest.java @@ -0,0 +1,34 @@ +package games.dmg.spigotevents.outbox; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import games.dmg.spigotevents.event.OutboundEvent; +import java.nio.file.Path; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +class SqliteOutboxTest { + @TempDir + Path temporaryDirectory; + + @Test + void retainsEventsAcrossRestartAndAcknowledgesOnlyTheSentPrefix() throws Exception { + Path database = temporaryDirectory.resolve("outbox.db"); + try (var outbox = new SqliteOutbox(database)) { + outbox.append(new OutboundEvent("one", "{\"id\":\"one\"}")); + outbox.append(new OutboundEvent("two", "{\"id\":\"two\"}")); + } + + try (var reopened = new SqliteOutbox(database)) { + var batch = reopened.peekBatch(1, 1_000); + assertEquals(1, batch.size()); + assertEquals("one", batch.get(0).eventId()); + assertEquals(2, reopened.pendingCount()); + + reopened.acknowledgeThrough(batch.get(0).sequence()); + + assertEquals(1, reopened.pendingCount()); + assertEquals("two", reopened.peekBatch(10, 1_000).get(0).eventId()); + } + } +} diff --git a/src/test/java/games/dmg/spigotevents/stats/MapChunkerTest.java b/src/test/java/games/dmg/spigotevents/stats/MapChunkerTest.java new file mode 100644 index 0000000..7079305 --- /dev/null +++ b/src/test/java/games/dmg/spigotevents/stats/MapChunkerTest.java @@ -0,0 +1,31 @@ +package games.dmg.spigotevents.stats; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.google.gson.Gson; +import java.nio.charset.StandardCharsets; +import java.util.LinkedHashMap; +import java.util.Map; +import org.junit.jupiter.api.Test; + +class MapChunkerTest { + @Test + void splitsLargeStatisticMapsWithoutDroppingValues() { + Map statistics = new LinkedHashMap<>(); + for (int index = 0; index < 100; index++) { + statistics.put("minecraft:test_statistic_" + index, index); + } + + var chunks = MapChunker.chunk(statistics, 250); + Map reconstructed = new LinkedHashMap<>(); + Gson gson = new Gson(); + for (var chunk : chunks) { + assertTrue(gson.toJson(chunk).getBytes(StandardCharsets.UTF_8).length <= 250); + reconstructed.putAll(chunk); + } + + assertTrue(chunks.size() > 1); + assertEquals(statistics, reconstructed); + } +}