PlaytimeHistory/Database.cs

using MySqlConnector;

namespace PlaytimeHistory;

// The plugin's own MySQL database (pt_* tables). Every method opens its own connection from the
// pool and runs off the game thread. Writes come only from the Writer, one at a time, in order.
public sealed class Database
{
    private const string MigrationPrefix = "PlaytimeHistory.Migrations.";

    private readonly string _connectionString;

    public Database(DatabaseConfig config)
    {
        if (!Enum.TryParse<MySqlSslMode>(config.SslMode, ignoreCase: true, out var sslMode))
            throw new ArgumentException($"Database.SslMode \"{config.SslMode}\" isn't one of None, Preferred, Required, VerifyCA, VerifyFull");

        _connectionString = new MySqlConnectionStringBuilder
        {
            Server = config.Host,
            Port = config.Port,
            Database = config.Database,
            UserID = config.User,
            Password = config.Password,
            SslMode = sslMode,
            ConnectionTimeout = 10,
            DefaultCommandTimeout = 30,
            MaximumPoolSize = 5,
        }.ConnectionString;
    }

    // What css_playtime shows. Totals come from pt_players; the week from closed segments.
    public sealed record Summary(
        DateTime FirstSeen, DateTime LastSeen, int Sessions, long Total, long T, long CT, long Spec,
        long Alive, long Dead, string LastNickname, long WeekPlaying, long WeekSpec);

    public sealed record RecentSession(DateTime StartedAt, DateTime? EndedAt, int? Seconds, string Map, EndKind EndKind);

    private async Task<MySqlConnection> OpenAsync(CancellationToken token = default)
    {
        var connection = new MySqlConnection(_connectionString);
        await connection.OpenAsync(token);
        return connection;
    }

    private static MySqlCommand Command(MySqlConnection connection, MySqlTransaction? transaction, string sql,
        params (string Name, object? Value)[] parameters)
    {
        var command = new MySqlCommand(sql, connection, transaction);
        foreach (var (name, value) in parameters)
            command.Parameters.AddWithValue(name, value ?? DBNull.Value);
        return command;
    }

    // Applies the embedded Migrations/*.sql in name order, each once, like GroupVerify's gv_migrations.
    // Returns the versions applied.
    public async Task<List<string>> MigrateAsync(CancellationToken token)
    {
        await using var connection = await OpenAsync(token);
        await using (var create = Command(connection, null, """
            CREATE TABLE IF NOT EXISTS `pt_migrations` (
                `id` int NOT NULL AUTO_INCREMENT,
                `version` varchar(128) NOT NULL,
                PRIMARY KEY (`id`)
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
            """))
            await create.ExecuteNonQueryAsync(token);

        string last;
        await using (var query = Command(connection, null, "SELECT version FROM pt_migrations ORDER BY id DESC LIMIT 1"))
            last = (await query.ExecuteScalarAsync(token))?.ToString() ?? "";

        var assembly = typeof(Database).Assembly;
        var applied = new List<string>();
        foreach (var resource in assembly.GetManifestResourceNames()
                     .Where(n => n.StartsWith(MigrationPrefix) && n.EndsWith(".sql"))
                     .Order(StringComparer.Ordinal))
        {
            var version = resource[MigrationPrefix.Length..^".sql".Length];
            if (string.CompareOrdinal(version, last) <= 0)
                continue;

            using var reader = new StreamReader(assembly.GetManifestResourceStream(resource)!);
            await using (var migrate = Command(connection, null, await reader.ReadToEndAsync(token)))
                await migrate.ExecuteNonQueryAsync(token);
            await using (var record = Command(connection, null, "INSERT INTO pt_migrations (version) VALUES (@version)", ("@version", version)))
                await record.ExecuteNonQueryAsync(token);
            applied.Add(version);
        }
        return applied;
    }

    public async Task ApplyAsync(Op op, CancellationToken token)
    {
        await using var connection = await OpenAsync(token);
        await using var transaction = await connection.BeginTransactionAsync(token);
        switch (op)
        {
            case ServerSeen o:
                await Exec(connection, transaction, token, """
                    INSERT INTO pt_servers (id, address, hostname, first_seen, last_seen)
                    VALUES (@id, @address, @hostname, @at, @at)
                    ON DUPLICATE KEY UPDATE
                        address = IF(VALUES(address) = '', address, VALUES(address)),
                        hostname = IF(VALUES(hostname) = '', hostname, VALUES(hostname)),
                        last_seen = GREATEST(last_seen, VALUES(last_seen))
                    """, ("@id", o.ServerId), ("@address", o.Address), ("@hostname", Clip(o.Hostname, 128)), ("@at", o.At));
                break;

            case RecoverStale o:
                await RecoverStaleAsync(connection, transaction, o, token);
                break;

            case SessionOpen o:
                await SessionOpenAsync(connection, transaction, o, token);
                break;

            case SessionClose o:
                await SessionCloseAsync(connection, transaction, o, token);
                break;

            case SegmentOpen o:
                await Exec(connection, transaction, token, """
                    INSERT IGNORE INTO pt_segments
                        (segment_key, session_key, steamid64, server_id, map, team, alive, started_at, heartbeat_at)
                    VALUES (@key, @session, @steamid, @server, @map, @team, @alive, @at, @at)
                    """, ("@key", o.SegmentKey), ("@session", o.SessionKey), ("@steamid", o.SteamId),
                    ("@server", o.ServerId), ("@map", Clip(o.Map, 64)), ("@team", o.Team), ("@alive", o.Alive), ("@at", o.StartedAt));
                break;

            case SegmentRestate o:
                await Exec(connection, transaction, token,
                    "UPDATE pt_segments SET team = @team, alive = @alive WHERE segment_key = @key AND ended_at IS NULL",
                    ("@key", o.SegmentKey), ("@team", o.Team), ("@alive", o.Alive));
                break;

            case SegmentClose o:
                await CloseSegmentAsync(connection, transaction, o.SegmentKey, o.EndedAt, o.CountSpectatorTime, token);
                break;

            case Heartbeat o:
                await Exec(connection, transaction, token,
                    "UPDATE pt_sessions SET heartbeat_at = @at WHERE server_id = @server AND ended_at IS NULL AND heartbeat_at < @at",
                    ("@server", o.ServerId), ("@at", o.At));
                await Exec(connection, transaction, token,
                    "UPDATE pt_segments SET heartbeat_at = @at WHERE server_id = @server AND ended_at IS NULL AND heartbeat_at < @at",
                    ("@server", o.ServerId), ("@at", o.At));
                break;

            case OnlineSnapshot o:
                await Exec(connection, transaction, token, """
                    INSERT INTO pt_online_snapshots (server_id, taken_at, players, bots, max_players, map)
                    VALUES (@server, @at, @players, @bots, @max, @map)
                    """, ("@server", o.ServerId), ("@at", o.At), ("@players", o.Players), ("@bots", o.Bots),
                    ("@max", o.MaxPlayers), ("@map", Clip(o.Map, 64)));
                break;

            case EventRecord o:
                await Exec(connection, transaction, token, """
                    INSERT IGNORE INTO pt_events
                        (event_key, server_id, session_key, steamid64, kind, at, map, team, weapon, headshot, noscope,
                         thrusmoke, wallbang, blind, victim_steamid64, victim_bot, site)
                    VALUES (@key, @server, @session, @steamid, @kind, @at, @map, @team, @weapon, @headshot, @noscope,
                         @thrusmoke, @wallbang, @blind, @victim, @victimBot, @site)
                    """, ("@key", o.EventKey), ("@server", o.ServerId), ("@session", o.SessionKey), ("@steamid", o.SteamId),
                    ("@kind", (int)o.Kind), ("@at", o.At), ("@map", Clip(o.Map, 64)), ("@team", o.Team),
                    ("@weapon", Clip(o.Weapon, 64)), ("@headshot", o.Headshot), ("@noscope", o.Noscope),
                    ("@thrusmoke", o.Thrusmoke), ("@wallbang", o.Wallbang), ("@blind", o.Blind),
                    ("@victim", o.VictimSteamId), ("@victimBot", o.VictimBot), ("@site", o.Site));
                break;

            default:
                throw new ArgumentException($"Unknown op {op.GetType().Name}");
        }
        await transaction.CommitAsync(token);
    }

    private static async Task SessionOpenAsync(MySqlConnection connection, MySqlTransaction transaction, SessionOpen o, CancellationToken token)
    {
        var nickname = Clip(o.Nickname, 128);
        var inserted = await Exec(connection, transaction, token, """
            INSERT IGNORE INTO pt_sessions
                (session_key, steamid64, server_id, nickname, started_at, heartbeat_at, connect_map,
                 players_online, max_players, country_iso, country_name, city, plugin_version)
            VALUES (@key, @steamid, @server, @nickname, @at, @at, @map, @online, @max, @iso, @country, @city, @version)
            """, ("@key", o.SessionKey), ("@steamid", o.SteamId), ("@server", o.ServerId), ("@nickname", nickname),
            ("@at", o.StartedAt), ("@map", Clip(o.Map, 64)), ("@online", o.PlayersOnline), ("@max", o.MaxPlayers),
            ("@iso", o.CountryIso), ("@country", Clip(o.CountryName, 64)), ("@city", Clip(o.City, 128)),
            ("@version", Clip(o.PluginVersion, 32)));
        if (inserted == 0)
            return;

        await Exec(connection, transaction, token, """
            INSERT INTO pt_players (steamid64, first_seen, last_seen, sessions_count, last_nickname, last_country, last_server_id)
            VALUES (@steamid, @at, @at, 1, @nickname, @iso, @server)
            ON DUPLICATE KEY UPDATE
                last_seen = GREATEST(last_seen, VALUES(last_seen)),
                sessions_count = sessions_count + 1,
                last_nickname = VALUES(last_nickname),
                last_country = COALESCE(VALUES(last_country), last_country),
                last_server_id = VALUES(last_server_id)
            """, ("@steamid", o.SteamId), ("@at", o.StartedAt), ("@nickname", nickname), ("@iso", o.CountryIso), ("@server", o.ServerId));

        await Exec(connection, transaction, token, """
            INSERT INTO pt_nicknames (steamid64, nickname, first_seen, last_seen, times_seen)
            VALUES (@steamid, @nickname, @at, @at, 1)
            ON DUPLICATE KEY UPDATE last_seen = GREATEST(last_seen, VALUES(last_seen)), times_seen = times_seen + 1
            """, ("@steamid", o.SteamId), ("@nickname", nickname), ("@at", o.StartedAt));
    }

    private static async Task SessionCloseAsync(MySqlConnection connection, MySqlTransaction transaction, SessionClose o, CancellationToken token)
    {
        var s = o.Stats;
        var p = o.Ping;
        var closed = await Exec(connection, transaction, token, """
            UPDATE pt_sessions SET
                ended_at = @at, heartbeat_at = @at,
                duration_seconds = GREATEST(TIMESTAMPDIFF(SECOND, started_at, @at), 0),
                end_kind = @kind, disconnect_map = @map, disconnect_reason = @reason, disconnect_reason_name = @reasonName,
                kills = @kills, deaths = @deaths, assists = @assists, headshots = @headshots, damage = @damage,
                mvp = @mvp, score = @score, rounds_played = @rounds,
                ping_avg = @pingAvg, ping_min = @pingMin, ping_max = @pingMax, ping_samples = @pingSamples
            WHERE session_key = @key AND ended_at IS NULL
            """, ("@key", o.SessionKey), ("@at", o.EndedAt), ("@kind", (int)o.EndKind), ("@map", Clip(o.Map, 64)),
            ("@reason", o.DisconnectReason), ("@reasonName", Clip(o.DisconnectReasonName, 64)),
            ("@kills", s?.Kills), ("@deaths", s?.Deaths), ("@assists", s?.Assists), ("@headshots", s?.Headshots),
            ("@damage", s?.Damage), ("@mvp", s?.Mvp), ("@score", s?.Score), ("@rounds", o.RoundsPlayed),
            ("@pingAvg", p?.Avg), ("@pingMin", p?.Min), ("@pingMax", p?.Max), ("@pingSamples", p?.Samples));
        if (closed == 0)
            return;

        await Exec(connection, transaction, token,
            "UPDATE pt_players SET last_seen = GREATEST(last_seen, @at) WHERE steamid64 = @steamid",
            ("@steamid", o.SteamId), ("@at", o.EndedAt));
    }

    // Closes a segment that's still open and adds its time to the player's totals, together.
    private static async Task CloseSegmentAsync(MySqlConnection connection, MySqlTransaction transaction,
        string segmentKey, DateTime endedAt, bool countSpectatorTime, CancellationToken token)
    {
        ulong steamId;
        int team;
        bool? alive;
        DateTime startedAt;
        await using (var select = Command(connection, transaction,
                         "SELECT steamid64, team, alive, started_at FROM pt_segments WHERE segment_key = @key AND ended_at IS NULL FOR UPDATE",
                         ("@key", segmentKey)))
        await using (var reader = await select.ExecuteReaderAsync(token))
        {
            if (!await reader.ReadAsync(token))
                return;
            steamId = reader.GetUInt64(0);
            team = reader.GetInt32(1);
            alive = reader.IsDBNull(2) ? null : reader.GetInt32(2) != 0;
            startedAt = reader.GetDateTime(3);
        }

        var seconds = Math.Max(0L, (long)(endedAt - startedAt).TotalSeconds);
        await Exec(connection, transaction, token,
            "UPDATE pt_segments SET ended_at = @at, heartbeat_at = @at, seconds = @seconds WHERE segment_key = @key",
            ("@key", segmentKey), ("@at", endedAt), ("@seconds", seconds));

        var playing = Teams.Playing(team);
        await Exec(connection, transaction, token, """
            UPDATE pt_players SET
                total_seconds = total_seconds + @total,
                t_seconds = t_seconds + @t,
                ct_seconds = ct_seconds + @ct,
                spec_seconds = spec_seconds + @spec,
                alive_seconds = alive_seconds + @alive,
                dead_seconds = dead_seconds + @dead
            WHERE steamid64 = @steamid
            """, ("@steamid", steamId),
            ("@total", playing || countSpectatorTime ? seconds : 0),
            ("@t", team == Teams.T ? seconds : 0),
            ("@ct", team == Teams.CT ? seconds : 0),
            ("@spec", playing ? 0 : seconds),
            ("@alive", playing && alive == true ? seconds : 0),
            ("@dead", playing && alive != true ? seconds : 0));
    }

    private static async Task RecoverStaleAsync(MySqlConnection connection, MySqlTransaction transaction, RecoverStale o, CancellationToken token)
    {
        var open = new List<(string Key, DateTime Heartbeat)>();
        await using (var select = Command(connection, transaction,
                         "SELECT segment_key, heartbeat_at FROM pt_segments WHERE server_id = @server AND ended_at IS NULL",
                         ("@server", o.ServerId)))
        await using (var reader = await select.ExecuteReaderAsync(token))
            while (await reader.ReadAsync(token))
                open.Add((reader.GetString(0), reader.GetDateTime(1)));

        foreach (var (key, heartbeat) in open)
            await CloseSegmentAsync(connection, transaction, key, heartbeat, o.CountSpectatorTime, token);

        await Exec(connection, transaction, token, """
            UPDATE pt_players p JOIN pt_sessions s ON s.steamid64 = p.steamid64
            SET p.last_seen = GREATEST(p.last_seen, s.heartbeat_at)
            WHERE s.server_id = @server AND s.ended_at IS NULL
            """, ("@server", o.ServerId));
        await Exec(connection, transaction, token, """
            UPDATE pt_sessions SET
                ended_at = heartbeat_at,
                duration_seconds = GREATEST(TIMESTAMPDIFF(SECOND, started_at, heartbeat_at), 0),
                end_kind = @kind
            WHERE server_id = @server AND ended_at IS NULL
            """, ("@server", o.ServerId), ("@kind", (int)EndKind.Stale));
    }

    public async Task<Summary?> GetSummaryAsync(ulong steamId, CancellationToken token)
    {
        await using var connection = await OpenAsync(token);
        Summary? summary = null;
        await using (var select = Command(connection, null, """
                         SELECT first_seen, last_seen, sessions_count, total_seconds, t_seconds, ct_seconds,
                                spec_seconds, alive_seconds, dead_seconds, last_nickname
                         FROM pt_players WHERE steamid64 = @steamid
                         """, ("@steamid", steamId)))
        await using (var reader = await select.ExecuteReaderAsync(token))
        {
            if (!await reader.ReadAsync(token))
                return null;
            summary = new Summary(reader.GetDateTime(0), reader.GetDateTime(1), reader.GetInt32(2), reader.GetInt64(3),
                reader.GetInt64(4), reader.GetInt64(5), reader.GetInt64(6), reader.GetInt64(7), reader.GetInt64(8),
                reader.GetString(9), 0, 0);
        }

        await using (var week = Command(connection, null, """
                         SELECT COALESCE(SUM(CASE WHEN team IN (2, 3) THEN seconds END), 0),
                                COALESCE(SUM(CASE WHEN team IN (0, 1) THEN seconds END), 0)
                         FROM pt_segments
                         WHERE steamid64 = @steamid AND ended_at IS NOT NULL AND started_at >= UTC_TIMESTAMP() - INTERVAL 7 DAY
                         """, ("@steamid", steamId)))
        await using (var reader = await week.ExecuteReaderAsync(token))
        {
            if (await reader.ReadAsync(token))
                summary = summary with { WeekPlaying = reader.GetInt64(0), WeekSpec = reader.GetInt64(1) };
        }
        return summary;
    }

    public async Task<List<RecentSession>> GetRecentSessionsAsync(ulong steamId, int limit, CancellationToken token)
    {
        await using var connection = await OpenAsync(token);
        await using var select = Command(connection, null, """
            SELECT started_at, ended_at, duration_seconds, connect_map, end_kind
            FROM pt_sessions WHERE steamid64 = @steamid
            ORDER BY started_at DESC LIMIT @limit
            """, ("@steamid", steamId), ("@limit", limit));
        await using var reader = await select.ExecuteReaderAsync(token);
        var sessions = new List<RecentSession>();
        while (await reader.ReadAsync(token))
            sessions.Add(new RecentSession(
                reader.GetDateTime(0),
                reader.IsDBNull(1) ? null : reader.GetDateTime(1),
                reader.IsDBNull(2) ? null : reader.GetInt32(2),
                reader.GetString(3),
                (EndKind)reader.GetInt32(4)));
        return sessions;
    }

    private static async Task<int> Exec(MySqlConnection connection, MySqlTransaction transaction, CancellationToken token,
        string sql, params (string Name, object? Value)[] parameters)
    {
        await using var command = Command(connection, transaction, sql, parameters);
        return await command.ExecuteNonQueryAsync(token);
    }

    [return: System.Diagnostics.CodeAnalysis.NotNullIfNotNull(nameof(value))]
    private static string? Clip(string? value, int length) =>
        value == null || value.Length <= length ? value : value[..length];
}