Implement historic message replay.

20 files changed, 456 insertions, 3 deletions

diff --git a/src/main/java/com/keuin/crosslink/messaging/history/HistoricMessageRecorder.java b/src/main/java/com/keuin/crosslink/messaging/history/HistoricMessageRecorder.java
index 2412bb4..ba18d83 100644
--- a/src/main/java/com/keuin/crosslink/messaging/history/HistoricMessageRecorder.java
+++ b/src/main/java/com/keuin/crosslink/messaging/history/HistoricMessageRecorder.java
@@ -9,7 +9,7 @@ import java.util.Objects;
 import java.util.stream.Collectors;
 
 public class HistoricMessageRecorder implements IHistoricMessageRecorder {
-    private final long ttlMillis;
+    private long ttlMillis;
     private final LinkedList<Pair<Long, IMessage>> que = new LinkedList<>();
     private final Object lock = new Object();
 
@@ -17,6 +17,10 @@ public class HistoricMessageRecorder implements IHistoricMessageRecorder {
         this.ttlMillis = ttlMillis;
     }
 
+    public HistoricMessageRecorder() {
+        this.ttlMillis = 0;
+    }
+
     private void clean() {
         for (var head = que.peek();
              head != null &&
@@ -27,9 +31,11 @@ public class HistoricMessageRecorder implements IHistoricMessageRecorder {
             que.removeFirst();
         }
     }
+
     /**
      * Add and memorize a message.
      * Note: this implementation is synchronized. Be caution if you requires a high performance.
+     *
      * @param message the message to save.
      */
     @Override
@@ -48,4 +54,14 @@ public class HistoricMessageRecorder implements IHistoricMessageRecorder {
             return que.stream().map(Pair::getV).collect(Collectors.toList());
         }
     }
+
+    @Override
+    public long getTTL() {
+        return ttlMillis;
+    }
+
+    @Override
+    public void setTTL(long ttl) {
+        this.ttlMillis = ttl;
+    }
 }
diff --git a/src/main/java/com/keuin/crosslink/messaging/history/IHistoricMessageRecorder.java b/src/main/java/com/keuin/crosslink/messaging/history/IHistoricMessageRecorder.java
index f5f245d..93a310a 100644
--- a/src/main/java/com/keuin/crosslink/messaging/history/IHistoricMessageRecorder.java
+++ b/src/main/java/com/keuin/crosslink/messaging/history/IHistoricMessageRecorder.java
@@ -22,4 +22,16 @@ public interface IHistoricMessageRecorder {
      * @return an {@link List} containing all recent messages in the time scope.
      */
     List<IMessage> getMessages();
+
+    /**
+     * Get message TTL.
+     * @return the message TTL milliseconds.
+     */
+    long getTTL();
+
+    /**
+     * Set message TTL. (a workaround to implement later initialization)
+     * @param ttl the message TTL milliseconds.
+     */
+    void setTTL(long ttl);
 }
diff --git a/src/main/java/com/keuin/crosslink/plugin/bungee/BungeeAccessor.java b/src/main/java/com/keuin/crosslink/plugin/bungee/BungeeAccessor.java
index 525f16b..f840137 100644
--- a/src/main/java/com/keuin/crosslink/plugin/bungee/BungeeAccessor.java
+++ b/src/main/java/com/keuin/crosslink/plugin/bungee/BungeeAccessor.java
@@ -8,10 +8,13 @@ import com.keuin.crosslink.messaging.endpoint.IEndpoint;
 import com.keuin.crosslink.messaging.endpoint.local.BungeeServerChatEndpoint;
 import com.keuin.crosslink.plugin.bungee.checker.BungeeServerStatusChecker;
 import com.keuin.crosslink.plugin.common.ICoreAccessor;
+import net.kyori.adventure.text.Component;
+import net.md_5.bungee.api.chat.ComponentBuilder;
 
 import java.util.List;
 import java.util.Objects;
 import java.util.Set;
+import java.util.UUID;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
@@ -55,4 +58,13 @@ public class BungeeAccessor implements ICoreAccessor {
                 .map((si) -> new BungeeServerChatEndpoint(si, plugin.getProxy(), plugin))
                 .collect(Collectors.toUnmodifiableSet());
     }
+
+    @Override
+    public void sendPlayerMessage(UUID playerUuid, Component message) {
+        var player = plugin.getProxy().getPlayer(playerUuid);
+        if (player == null) return;
+        // FIXME keep color data
+        var msg = new ComponentBuilder().append(message.toString()).create();
+        player.sendMessage(msg);
+    }
 }
diff --git a/src/main/java/com/keuin/crosslink/plugin/bungee/BungeeEventBus.java b/src/main/java/com/keuin/crosslink/plugin/bungee/BungeeEventBus.java
new file mode 100644
index 0000000..4ccff85
--- /dev/null
+++ b/src/main/java/com/keuin/crosslink/plugin/bungee/BungeeEventBus.java
@@ -0,0 +1,155 @@
+package com.keuin.crosslink.plugin.bungee;
+
+import com.google.inject.Inject;
+import com.keuin.crosslink.plugin.common.IEventBus;
+import com.keuin.crosslink.plugin.common.event.*;
+import net.md_5.bungee.api.ProxyServer;
+import net.md_5.bungee.api.config.ServerInfo;
+import net.md_5.bungee.api.connection.ProxiedPlayer;
+import net.md_5.bungee.api.event.PlayerDisconnectEvent;
+import net.md_5.bungee.api.event.*;
+import net.md_5.bungee.api.plugin.Listener;
+import net.md_5.bungee.api.plugin.Plugin;
+import org.jetbrains.annotations.NotNull;
+
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.logging.Logger;
+
+public class BungeeEventBus implements Listener, IEventBus {
+    private final Plugin plugin;
+    private final Logger logger = Logger.getLogger(BungeeEventBus.class.getName());
+    private final Set<UUID> connectedPlayers = Collections.newSetFromMap(new ConcurrentHashMap<>()); // all players connected to the proxy
+    private final Map<UUID, ServerInfo> joiningServers = new HashMap<>();
+    private final Map<UUID, ServerInfo> serverPlayerLastJoined = new HashMap<>(); // the server players last connected to
+
+    private final List<EventHandler> handlers = new ArrayList<>();
+
+    @Inject
+    public BungeeEventBus(Plugin plugin) {
+        this.plugin = Objects.requireNonNull(plugin);
+    }
+
+    @Override
+    public void registerEventHandler(@NotNull com.keuin.crosslink.plugin.common.event.EventHandler handler) {
+        Objects.requireNonNull(handler);
+        handlers.add(handler);
+    }
+
+    @net.md_5.bungee.event.EventHandler
+    public void onServerDisconnect(ServerDisconnectEvent event) {
+        ProxyServer.getInstance().getScheduler().runAsync(plugin, () -> {
+            for (var handler : handlers) {
+                if (handler instanceof PlayerQuitServerEvent) {
+                    ((PlayerQuitServerEvent) handler).onPlayerQuitServer(
+                            event.getPlayer().getName(), event.getTarget().getName());
+                }
+            }
+        });
+    }
+
+    @net.md_5.bungee.event.EventHandler
+    public void onPlayerDisconnect(PlayerDisconnectEvent event) {
+        try {
+            var player = event.getPlayer();
+            if (player != null) {
+                var playerId = player.getName();
+                var playerUuid = player.getUniqueId();
+                var server = Optional
+                        .ofNullable(serverPlayerLastJoined.get(player.getUniqueId()))
+                        .map(ServerInfo::getName)
+                        .orElse("<unknown server>"); // this should not happen
+                logger.info("Player " + playerId + " disconnected. Remove him from local online list.");
+                connectedPlayers.remove(playerUuid);
+                ProxyServer.getInstance().getScheduler().runAsync(plugin, () -> {
+                    for (var handler : handlers) {
+                        if (handler instanceof com.keuin.crosslink.plugin.common.event.PlayerDisconnectEvent) {
+                            ((com.keuin.crosslink.plugin.common.event.PlayerDisconnectEvent) handler)
+                                    .onPlayerDisconnect(playerId, server);
+                        }
+                    }
+                });
+            }
+        } catch (Exception e) {
+            logger.warning(String.format("An exception caught while handling player disconnect event: %s.", e));
+        }
+    }
+
+    @net.md_5.bungee.event.EventHandler
+    public void onServerConnect(ServerConnectEvent event) {
+        ProxiedPlayer player = event.getPlayer();
+        if (player == null)
+            return;
+        ServerInfo server = event.getTarget();
+        joiningServers.put(player.getUniqueId(), server);
+    }
+
+    @net.md_5.bungee.event.EventHandler
+    public void onPlayerJoined(ServerConnectedEvent event) {
+
+        // after a player has joined the server
+        ProxiedPlayer player = event.getPlayer();
+        if (player == null)
+            return;
+        ProxyServer proxy = plugin.getProxy();
+        ServerInfo server = joiningServers.get(player.getUniqueId());
+
+        if (!joiningServers.containsKey(event.getPlayer().getUniqueId())) {
+            logger.warning(String.format(
+                    "Unexpected player %s. Login broadcast will not be sent.",
+                    event.getPlayer().getName()));
+            return;
+        }
+
+        ProxyServer.getInstance().getScheduler().schedule(plugin, () -> {
+            for (var handler : handlers) {
+                if (handler instanceof PlayerJoinServerEvent) {
+                    ((PlayerJoinServerEvent) handler)
+                            .onPlayerJoinServer(player.getName(), server.getName());
+                }
+            }
+        }, 250, TimeUnit.MILLISECONDS); // to make the message sequence correct: a stupid work-around
+
+        joiningServers.remove(event.getPlayer().getUniqueId());
+        serverPlayerLastJoined.put(event.getPlayer().getUniqueId(), server);
+
+        if (connectedPlayers.add(player.getUniqueId())) {
+            // The BungeeCord provides a very tedious event API, so we have to work around like this.
+            // Otherwise, the client won't receive any message.
+            ProxyServer.getInstance().getScheduler().schedule(plugin, () -> {
+                logger.info("Player " + player.getName() + " logged in." +
+                        " Add him to local login list and send him recent messages.");
+                for (var handler : handlers) {
+                    if (handler instanceof PlayerConnectEvent) {
+                        ((PlayerConnectEvent) handler)
+                                .onPlayerConnect(player.getName(), player.getUniqueId());
+                    }
+                }
+            }, 100, TimeUnit.MILLISECONDS);
+        }
+    }
+
+    public void onPlayerChat(ChatEvent event) {
+        var conn = event.getSender();
+        if (!(conn instanceof ProxiedPlayer)) {
+            logger.severe(String.format("Sender is not a ProxiedPlayer instance: %s", event.getSender().toString()));
+            return;
+        }
+
+        var sender = ((ProxiedPlayer) conn).getDisplayName();
+        var server = ((ProxiedPlayer) conn).getServer().getInfo().getName();
+        var message = event.getMessage();
+
+        if (message.startsWith("/"))
+            return; // Do not repeat commands
+
+        logger.info(String.format("Chat message: %s, sender: %s", message, sender));
+
+        for (var handler : handlers) {
+            if (handler instanceof PlayerChatEvent) {
+                ((PlayerChatEvent) handler).onPlayerChat(sender, server, message);
+            }
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/main/java/com/keuin/crosslink/plugin/bungee/BungeeMainWrapper.java b/src/main/java/com/keuin/crosslink/plugin/bungee/BungeeMainWrapper.java
index e3f9824..294e574 100644
--- a/src/main/java/com/keuin/crosslink/plugin/bungee/BungeeMainWrapper.java
+++ b/src/main/java/com/keuin/crosslink/plugin/bungee/BungeeMainWrapper.java
@@ -4,10 +4,12 @@ import com.google.inject.Guice;
 import com.google.inject.Injector;
 import com.keuin.crosslink.plugin.bungee.module.BungeeAccessorModule;
 import com.keuin.crosslink.plugin.bungee.module.BungeeApiServerModule;
+import com.keuin.crosslink.plugin.bungee.module.BungeeEventBusModule;
 import com.keuin.crosslink.plugin.common.PluginMain;
 import com.keuin.crosslink.plugin.common.ProxyType;
 import com.keuin.crosslink.plugin.common.environ.PluginEnvironment;
 import com.keuin.crosslink.plugin.common.module.CommonApiServerProvider;
+import com.keuin.crosslink.plugin.common.module.CommonHistoricMessageRecorderModule;
 import com.keuin.crosslink.plugin.common.module.CommonIRouterModule;
 import com.keuin.crosslink.plugin.common.module.CommonPluginEnvironProvider;
 import com.keuin.crosslink.util.LoggerNaming;
@@ -19,6 +21,8 @@ public final class BungeeMainWrapper extends Plugin {
     private final Injector injector = Guice.createInjector(
             new BungeeAccessorModule(this),
             new BungeeApiServerModule(),
+            new BungeeEventBusModule(),
+            new CommonHistoricMessageRecorderModule(),
             new CommonIRouterModule(),
             new CommonPluginEnvironProvider(new PluginEnvironment(
                     ProxyType.BUNGEECORD,
diff --git a/src/main/java/com/keuin/crosslink/plugin/bungee/module/BungeeEventBusModule.java b/src/main/java/com/keuin/crosslink/plugin/bungee/module/BungeeEventBusModule.java
new file mode 100644
index 0000000..e8c54f7
--- /dev/null
+++ b/src/main/java/com/keuin/crosslink/plugin/bungee/module/BungeeEventBusModule.java
@@ -0,0 +1,12 @@
+package com.keuin.crosslink.plugin.bungee.module;
+
+import com.google.inject.AbstractModule;
+import com.keuin.crosslink.plugin.bungee.BungeeEventBus;
+import com.keuin.crosslink.plugin.common.IEventBus;
+
+public class BungeeEventBusModule extends AbstractModule {
+    @Override
+    protected void configure() {
+        bind(IEventBus.class).to(BungeeEventBus.class);
+    }
+}
\ No newline at end of file
diff --git a/src/main/java/com/keuin/crosslink/plugin/common/ICoreAccessor.java b/src/main/java/com/keuin/crosslink/plugin/common/ICoreAccessor.java
index 4f0f0e0..77b83a0 100644
--- a/src/main/java/com/keuin/crosslink/plugin/common/ICoreAccessor.java
+++ b/src/main/java/com/keuin/crosslink/plugin/common/ICoreAccessor.java
@@ -3,9 +3,11 @@ package com.keuin.crosslink.plugin.common;
 import com.keuin.crosslink.data.PlayerInfo;
 import com.keuin.crosslink.data.ServerInfo;
 import com.keuin.crosslink.messaging.endpoint.IEndpoint;
+import net.kyori.adventure.text.Component;
 
 import java.util.List;
 import java.util.Set;
+import java.util.UUID;
 import java.util.function.Consumer;
 
 public interface ICoreAccessor {
@@ -16,4 +18,6 @@ public interface ICoreAccessor {
     void getServerInfo(Consumer<List<ServerInfo>> callback);
 
     Set<IEndpoint> getServerEndpoints();
+
+    void sendPlayerMessage(UUID playerUuid, Component message);
 }
diff --git a/src/main/java/com/keuin/crosslink/plugin/common/IEventBus.java b/src/main/java/com/keuin/crosslink/plugin/common/IEventBus.java
new file mode 100644
index 0000000..8c92535
--- /dev/null
+++ b/src/main/java/com/keuin/crosslink/plugin/common/IEventBus.java
@@ -0,0 +1,11 @@
+package com.keuin.crosslink.plugin.common;
+
+import com.keuin.crosslink.plugin.common.event.EventHandler;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Provides framework-independent event registration for CrossLink's upper-level logic.
+ */
+public interface IEventBus {
+    void registerEventHandler(@NotNull EventHandler handler);
+}
diff --git a/src/main/java/com/keuin/crosslink/plugin/common/PluginMain.java b/src/main/java/com/keuin/crosslink/plugin/common/PluginMain.java
index dff53a5..b0e5153 100644
--- a/src/main/java/com/keuin/crosslink/plugin/common/PluginMain.java
+++ b/src/main/java/com/keuin/crosslink/plugin/common/PluginMain.java
@@ -13,8 +13,10 @@ import com.keuin.crosslink.messaging.config.remote.RemoteEndpointFactory;
 import com.keuin.crosslink.messaging.config.router.RouterConfigurer;
 import com.keuin.crosslink.messaging.endpoint.IEndpoint;
 import com.keuin.crosslink.messaging.endpoint.system.ApiEndpoint;
+import com.keuin.crosslink.messaging.history.IHistoricMessageRecorder;
 import com.keuin.crosslink.messaging.router.IRouter;
 import com.keuin.crosslink.plugin.common.environ.PluginEnvironment;
+import com.keuin.crosslink.plugin.common.event.PlayerConnectEvent;
 import com.keuin.crosslink.util.LoggerNaming;
 import com.keuin.crosslink.util.StartupMessagePrinter;
 import com.keuin.crosslink.util.version.NewVersionChecker;
@@ -35,15 +37,21 @@ public final class PluginMain {
     private final IApiServer apiServer;
     private final ICoreAccessor coreAccessor;
     private final IRouter messageRouter;
+    private final IEventBus eventBus;
+    private final IHistoricMessageRecorder historicMessageRecorder;
     private final Logger logger;
 
     @Inject
     public PluginMain(ICoreAccessor coreAccessor,
                       IRouter messageRouter,
                       IApiServer apiServer,
-                      PluginEnvironment pluginEnvironment) {
+                      PluginEnvironment pluginEnvironment,
+                      IEventBus eventBus,
+                      IHistoricMessageRecorder historicMessageRecorder) {
         this.coreAccessor = coreAccessor;
         this.messageRouter = messageRouter;
+        this.eventBus = eventBus;
+        this.historicMessageRecorder = historicMessageRecorder;
         this.apiServer = apiServer;
         this.environment = pluginEnvironment;
         this.logger = environment.logger();
@@ -60,14 +68,30 @@ public final class PluginMain {
         // initialize message routing
         logger.info("Initializing message routing.");
         // Contains all enabled endpoints, including local and remote ones. Remote endpoints will be added later.
-        final var endpoints = new HashSet<>(coreAccessor.getServerEndpoints());
+        final var endpoints = new HashSet<IEndpoint>();
         try {
             var messaging = GlobalConfigManager.getInstance().messaging();
+            var local = Optional.ofNullable(messaging.get("local"))
+                    .orElse(mapper.readTree("{}"));
             var routing = Optional.ofNullable(messaging.get("routing"))
                     .orElse(mapper.readTree("[]"));
             var remote = Optional.ofNullable(messaging.get("remotes"))
                     .orElse(mapper.readTree("[]"));
 
+            // load local server endpoints
+            final var locals = coreAccessor.getServerEndpoints();
+            final var ttlSeconds = Optional.ofNullable(local.get("message_playback_seconds"))
+                    .map(JsonNode::asLong).orElse(-1L);
+            if (ttlSeconds > 0) {
+                // enable message replay
+                logger.info("Message replay is enabled. TTL: ");
+                eventBus.registerEventHandler(
+                        (PlayerConnectEvent) (player, uuid) -> historicMessageRecorder.getMessages()
+                                .forEach(msg -> coreAccessor.sendPlayerMessage(uuid, msg.kyoriMessage()))
+                );
+            }
+            endpoints.addAll(locals);
+
             // load routing table
             try {
                 logger.debug("Loading rule chain.");
diff --git a/src/main/java/com/keuin/crosslink/plugin/common/event/EventHandler.java b/src/main/java/com/keuin/crosslink/plugin/common/event/EventHandler.java
new file mode 100644
index 0000000..0226cea
--- /dev/null
+++ b/src/main/java/com/keuin/crosslink/plugin/common/event/EventHandler.java
@@ -0,0 +1,4 @@
+package com.keuin.crosslink.plugin.common.event;
+
+public interface EventHandler {
+}
diff --git a/src/main/java/com/keuin/crosslink/plugin/common/event/PlayerChatEvent.java b/src/main/java/com/keuin/crosslink/plugin/common/event/PlayerChatEvent.java
new file mode 100644
index 0000000..4838949
--- /dev/null
+++ b/src/main/java/com/keuin/crosslink/plugin/common/event/PlayerChatEvent.java
@@ -0,0 +1,6 @@
+package com.keuin.crosslink.plugin.common.event;
+
+@Deprecated
+public interface PlayerChatEvent extends EventHandler {
+    void onPlayerChat(String player, String server, String message);
+}
diff --git a/src/main/java/com/keuin/crosslink/plugin/common/event/PlayerConnectEvent.java b/src/main/java/com/keuin/crosslink/plugin/common/event/PlayerConnectEvent.java
new file mode 100644
index 0000000..5ecf3f2
--- /dev/null
+++ b/src/main/java/com/keuin/crosslink/plugin/common/event/PlayerConnectEvent.java
@@ -0,0 +1,12 @@
+package com.keuin.crosslink.plugin.common.event;
+
+import org.jetbrains.annotations.NotNull;
+
+import java.util.UUID;
+
+/**
+ * When a player connects to the proxy server.
+ */
+public interface PlayerConnectEvent extends EventHandler {
+    void onPlayerConnect(@NotNull String player, @NotNull UUID uuid);
+}
diff --git a/src/main/java/com/keuin/crosslink/plugin/common/event/PlayerDisconnectEvent.java b/src/main/java/com/keuin/crosslink/plugin/common/event/PlayerDisconnectEvent.java
new file mode 100644
index 0000000..941f3d8
--- /dev/null
+++ b/src/main/java/com/keuin/crosslink/plugin/common/event/PlayerDisconnectEvent.java
@@ -0,0 +1,10 @@
+package com.keuin.crosslink.plugin.common.event;
+
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * When a player disconnects from the proxy server.
+ */
+public interface PlayerDisconnectEvent extends EventHandler {
+    void onPlayerDisconnect(@NotNull String player, @NotNull String server);
+}
diff --git a/src/main/java/com/keuin/crosslink/plugin/common/event/PlayerJoinServerEvent.java b/src/main/java/com/keuin/crosslink/plugin/common/event/PlayerJoinServerEvent.java
new file mode 100644
index 0000000..e3b10c4
--- /dev/null
+++ b/src/main/java/com/keuin/crosslink/plugin/common/event/PlayerJoinServerEvent.java
@@ -0,0 +1,10 @@
+package com.keuin.crosslink.plugin.common.event;
+
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * When a player joins a sub-server.
+ */
+public interface PlayerJoinServerEvent extends EventHandler {
+    void onPlayerJoinServer(@NotNull String player, @NotNull String server);
+}
diff --git a/src/main/java/com/keuin/crosslink/plugin/common/event/PlayerQuitServerEvent.java b/src/main/java/com/keuin/crosslink/plugin/common/event/PlayerQuitServerEvent.java
new file mode 100644
index 0000000..d1f25dc
--- /dev/null
+++ b/src/main/java/com/keuin/crosslink/plugin/common/event/PlayerQuitServerEvent.java
@@ -0,0 +1,10 @@
+package com.keuin.crosslink.plugin.common.event;
+
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * When a player quits a sub-server.
+ */
+public interface PlayerQuitServerEvent extends EventHandler {
+    void onPlayerQuitServer(@NotNull String player, @NotNull String server);
+}
diff --git a/src/main/java/com/keuin/crosslink/plugin/common/module/CommonHistoricMessageRecorderModule.java b/src/main/java/com/keuin/crosslink/plugin/common/module/CommonHistoricMessageRecorderModule.java
new file mode 100644
index 0000000..6222735
--- /dev/null
+++ b/src/main/java/com/keuin/crosslink/plugin/common/module/CommonHistoricMessageRecorderModule.java
@@ -0,0 +1,12 @@
+package com.keuin.crosslink.plugin.common.module;
+
+import com.google.inject.AbstractModule;
+import com.keuin.crosslink.messaging.history.HistoricMessageRecorder;
+import com.keuin.crosslink.messaging.history.IHistoricMessageRecorder;
+
+public class CommonHistoricMessageRecorderModule extends AbstractModule {
+    @Override
+    protected void configure() {
+        bind(IHistoricMessageRecorder.class).to(HistoricMessageRecorder.class);
+    }
+}
diff --git a/src/main/java/com/keuin/crosslink/plugin/velocity/VelocityAccessor.java b/src/main/java/com/keuin/crosslink/plugin/velocity/VelocityAccessor.java
index f541139..7f76a90 100644
--- a/src/main/java/com/keuin/crosslink/plugin/velocity/VelocityAccessor.java
+++ b/src/main/java/com/keuin/crosslink/plugin/velocity/VelocityAccessor.java
@@ -7,9 +7,11 @@ import com.keuin.crosslink.messaging.endpoint.IEndpoint;
 import com.keuin.crosslink.messaging.endpoint.local.VelocityServerChatEndpoint;
 import com.keuin.crosslink.plugin.common.ICoreAccessor;
 import com.keuin.crosslink.plugin.velocity.checker.VelocityServerStatusChecker;
+import net.kyori.adventure.text.Component;
 
 import java.util.List;
 import java.util.Set;
+import java.util.UUID;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
@@ -51,4 +53,11 @@ public class VelocityAccessor implements ICoreAccessor {
                 .map((si) -> new VelocityServerChatEndpoint(si, plugin.getProxy(), plugin))
                 .collect(Collectors.toUnmodifiableSet());
     }
+
+    @Override
+    public void sendPlayerMessage(UUID playerUuid, Component message) {
+        var player = plugin.getProxy().getPlayer(playerUuid).orElse(null);
+        if (player == null) return;
+        player.sendMessage(message);
+    }
 }
diff --git a/src/main/java/com/keuin/crosslink/plugin/velocity/VelocityEventBus.java b/src/main/java/com/keuin/crosslink/plugin/velocity/VelocityEventBus.java
new file mode 100644
index 0000000..e47d9f2
--- /dev/null
+++ b/src/main/java/com/keuin/crosslink/plugin/velocity/VelocityEventBus.java
@@ -0,0 +1,114 @@
+package com.keuin.crosslink.plugin.velocity;
+
+import com.google.inject.Inject;
+import com.keuin.crosslink.plugin.bungee.BungeeEventBus;
+import com.keuin.crosslink.plugin.common.IEventBus;
+import com.keuin.crosslink.plugin.common.event.EventHandler;
+import com.keuin.crosslink.plugin.common.event.PlayerConnectEvent;
+import com.keuin.crosslink.plugin.common.event.PlayerJoinServerEvent;
+import com.keuin.crosslink.plugin.common.event.PlayerQuitServerEvent;
+import com.velocitypowered.api.event.Subscribe;
+import com.velocitypowered.api.event.connection.DisconnectEvent;
+import com.velocitypowered.api.event.connection.PostLoginEvent;
+import com.velocitypowered.api.event.player.ServerConnectedEvent;
+import com.velocitypowered.api.plugin.Plugin;
+import com.velocitypowered.api.proxy.ProxyServer;
+import com.velocitypowered.api.proxy.server.ServerInfo;
+import org.jetbrains.annotations.NotNull;
+
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.logging.Logger;
+
+public class VelocityEventBus implements IEventBus {
+    private final Plugin plugin;
+    private final ProxyServer velocity;
+    private final Logger logger = Logger.getLogger(BungeeEventBus.class.getName());
+    private final Set<UUID> connectedPlayers = Collections.newSetFromMap(new ConcurrentHashMap<>()); // all players connected to the proxy
+    private final Map<UUID, ServerInfo> serverPlayerLastJoined = new HashMap<>(); // the server players last connected to
+
+    private final List<EventHandler> handlers = new ArrayList<>();
+
+    @Inject
+    public VelocityEventBus(Plugin plugin, ProxyServer velocity) {
+        this.plugin = Objects.requireNonNull(plugin);
+        this.velocity = Objects.requireNonNull(velocity);
+    }
+
+    @Override
+    public void registerEventHandler(@NotNull com.keuin.crosslink.plugin.common.event.EventHandler handler) {
+        Objects.requireNonNull(handler);
+        handlers.add(handler);
+    }
+
+    @Subscribe
+    public void onPlayerConnect(PostLoginEvent event) {
+        var player = event.getPlayer();
+        var playerId = player.getUsername();
+        var playerUuid = player.getUniqueId();
+        velocity.getScheduler().buildTask(plugin, () -> {
+            for (var handler : handlers) {
+                if (handler instanceof PlayerConnectEvent) {
+                    ((PlayerConnectEvent) handler)
+                            .onPlayerConnect(playerId, playerUuid);
+                }
+            }
+        }).schedule();
+    }
+
+    @Subscribe
+    public void onPlayerDisconnect(DisconnectEvent event) {
+        try {
+            var player = event.getPlayer();
+            if (player != null) {
+                var playerId = player.getUsername();
+                var playerUuid = player.getUniqueId();
+                var server = Optional
+                        .ofNullable(serverPlayerLastJoined.get(player.getUniqueId()))
+                        .map(ServerInfo::getName)
+                        .orElse(null); // if this is null, the player does not connect to a server yet
+                logger.info("Player " + playerId + " disconnected. Remove him from local online list.");
+                connectedPlayers.remove(playerUuid);
+                if (server != null) {
+                    // also disconnect from a server
+                    velocity.getScheduler().buildTask(plugin, () -> {
+                        for (var handler : handlers) {
+                            if (handler instanceof PlayerQuitServerEvent) {
+                                ((PlayerQuitServerEvent) handler)
+                                        .onPlayerQuitServer(playerId, server);
+                            }
+                        }
+                    }).schedule();
+                }
+                // disconnect from the proxy
+                velocity.getScheduler().buildTask(plugin, () -> {
+                    for (var handler : handlers) {
+                        if (handler instanceof com.keuin.crosslink.plugin.common.event.PlayerDisconnectEvent) {
+                            ((com.keuin.crosslink.plugin.common.event.PlayerDisconnectEvent) handler)
+                                    .onPlayerDisconnect(playerId, Optional.ofNullable(server).orElse("<no server>"));
+                        }
+                    }
+                }).schedule();
+            }
+        } catch (Exception e) {
+            logger.warning(String.format("An exception caught while handling player disconnect event: %s.", e));
+        }
+    }
+
+    @Subscribe
+    public void onPlayerJoinedFinish(ServerConnectedEvent event) {
+        var player = event.getPlayer();
+        var playerId = player.getUsername();
+        var serverInfo = event.getServer().getServerInfo();
+        var server = serverInfo.getName();
+        serverPlayerLastJoined.put(player.getUniqueId(), serverInfo);
+        velocity.getScheduler().buildTask(plugin, () -> {
+            for (var handler : handlers) {
+                if (handler instanceof PlayerJoinServerEvent) {
+                    ((PlayerJoinServerEvent) handler)
+                            .onPlayerJoinServer(playerId, server);
+                }
+            }
+        }).schedule();
+    }
+}
\ No newline at end of file
diff --git a/src/main/java/com/keuin/crosslink/plugin/velocity/VelocityMainWrapper.java b/src/main/java/com/keuin/crosslink/plugin/velocity/VelocityMainWrapper.java
index f6a7cc5..9a1b177 100644
--- a/src/main/java/com/keuin/crosslink/plugin/velocity/VelocityMainWrapper.java
+++ b/src/main/java/com/keuin/crosslink/plugin/velocity/VelocityMainWrapper.java
@@ -6,10 +6,12 @@ import com.keuin.crosslink.plugin.common.PluginMain;
 import com.keuin.crosslink.plugin.common.ProxyType;
 import com.keuin.crosslink.plugin.common.environ.PluginEnvironment;
 import com.keuin.crosslink.plugin.common.module.CommonApiServerProvider;
+import com.keuin.crosslink.plugin.common.module.CommonHistoricMessageRecorderModule;
 import com.keuin.crosslink.plugin.common.module.CommonIRouterModule;
 import com.keuin.crosslink.plugin.common.module.CommonPluginEnvironProvider;
 import com.keuin.crosslink.plugin.velocity.module.VelocityAccessorModule;
 import com.keuin.crosslink.plugin.velocity.module.VelocityApiServerModule;
+import com.keuin.crosslink.plugin.velocity.module.VelocityEventBusModule;
 import com.keuin.crosslink.util.LoggerNaming;
 import com.velocitypowered.api.event.Subscribe;
 import com.velocitypowered.api.event.proxy.ProxyInitializeEvent;
@@ -48,6 +50,8 @@ public final class VelocityMainWrapper {
         var injector = Guice.createInjector(
                 new VelocityAccessorModule(this),
                 new VelocityApiServerModule(),
+                new VelocityEventBusModule(),
+                new CommonHistoricMessageRecorderModule(),
                 new CommonIRouterModule(),
                 new CommonPluginEnvironProvider(new PluginEnvironment(
                         ProxyType.VELOCITY, LoggerFactory.getLogger(LoggerNaming.name().toString()), pluginDataPath)),
diff --git a/src/main/java/com/keuin/crosslink/plugin/velocity/module/VelocityEventBusModule.java b/src/main/java/com/keuin/crosslink/plugin/velocity/module/VelocityEventBusModule.java
new file mode 100644
index 0000000..c0b90d7
--- /dev/null
+++ b/src/main/java/com/keuin/crosslink/plugin/velocity/module/VelocityEventBusModule.java
@@ -0,0 +1,12 @@
+package com.keuin.crosslink.plugin.velocity.module;
+
+import com.google.inject.AbstractModule;
+import com.keuin.crosslink.plugin.common.IEventBus;
+import com.keuin.crosslink.plugin.velocity.VelocityEventBus;
+
+public class VelocityEventBusModule extends AbstractModule {
+    @Override
+    protected void configure() {
+        bind(IEventBus.class).to(VelocityEventBus.class);
+    }
+}
\ No newline at end of file