liz/PluralKit
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor: add SqlKata for SQL generation, move connection handling into ModelRepository

Browse Source
This commit is contained in:
spiral
2021-09-29 21:51:38 -04:00
parent 6251d29abb
commit 92e45a07ff
60 changed files with 806 additions and 640 deletions
Download Patch File Download Diff File Expand all files Collapse all files

79
PluralKit.Core/Database/Database.cs
View File

@@ -2,6 +2,7 @@ using System;
using System.Collections.Generic;
using System.Data;
using System.IO;
using System.Runtime.CompilerServices;
using System.Threading.Tasks;
using App.Metrics;
@@ -14,6 +15,9 @@ using Npgsql;
using Serilog;
using SqlKata;
using SqlKata.Compilers;
namespace PluralKit.Core
{
internal class Database: IDatabase
@@ -46,6 +50,8 @@ namespace PluralKit.Core
}.ConnectionString;
}
private static readonly PostgresCompiler _compiler = new();
public static void InitStatic()
{
DefaultTypeMap.MatchNamesWithUnderscores = true;
@@ -151,28 +157,83 @@ namespace PluralKit.Core
public override T[] Parse(object value) => Array.ConvertAll((TInner[])value, v => _factory(v));
}
}
public static class DatabaseExt
{
public static async Task Execute(this IDatabase db, Func<IPKConnection, Task> func)
public async Task Execute(Func<IPKConnection, Task> func)
{
await using var conn = await db.Obtain();
await using var conn = await Obtain();
await func(conn);
}
public static async Task<T> Execute<T>(this IDatabase db, Func<IPKConnection, Task<T>> func)
public async Task<T> Execute<T>(Func<IPKConnection, Task<T>> func)
{
await using var conn = await db.Obtain();
await using var conn = await Obtain();
return await func(conn);
}
public static async IAsyncEnumerable<T> Execute<T>(this IDatabase db, Func<IPKConnection, IAsyncEnumerable<T>> func)
public async IAsyncEnumerable<T> Execute<T>(Func<IPKConnection, IAsyncEnumerable<T>> func)
{
await using var conn = await db.Obtain();
await using var conn = await Obtain();
await foreach (var val in func(conn))
yield return val;
}
public async Task<int> ExecuteQuery(Query q, string extraSql = "", [CallerMemberName] string queryName = "")
{
var query = _compiler.Compile(q);
using var conn = await Obtain();
using (_metrics.Measure.Timer.Time(CoreMetrics.DatabaseQuery, new MetricTags("Query", queryName)))
return await conn.ExecuteAsync(query.Sql + $" {extraSql}", query.NamedBindings);
}
public async Task<T> QueryFirst<T>(Query q, string extraSql = "", [CallerMemberName] string queryName = "")
{
var query = _compiler.Compile(q);
using var conn = await Obtain();
using (_metrics.Measure.Timer.Time(CoreMetrics.DatabaseQuery, new MetricTags("Query", queryName)))
return await conn.QueryFirstOrDefaultAsync<T>(query.Sql + $" {extraSql}", query.NamedBindings);
}
public async Task<T> QueryFirst<T>(IPKConnection? conn, Query q, string extraSql = "", [CallerMemberName] string queryName = "")
{
if (conn == null)
return await QueryFirst<T>(q, extraSql, queryName);
var query = _compiler.Compile(q);
using (_metrics.Measure.Timer.Time(CoreMetrics.DatabaseQuery, new MetricTags("Query", queryName)))
return await conn.QueryFirstOrDefaultAsync<T>(query.Sql + $" {extraSql}", query.NamedBindings);
}
public async Task<IEnumerable<T>> Query<T>(Query q, [CallerMemberName] string queryName = "")
{
var query = _compiler.Compile(q);
using var conn = await Obtain();
using (_metrics.Measure.Timer.Time(CoreMetrics.DatabaseQuery, new MetricTags("Query", queryName)))
return await conn.QueryAsync<T>(query.Sql, query.NamedBindings);
}
public async IAsyncEnumerable<T> QueryStream<T>(Query q, [CallerMemberName] string queryName = "")
{
var query = _compiler.Compile(q);
using var conn = await Obtain();
using (_metrics.Measure.Timer.Time(CoreMetrics.DatabaseQuery, new MetricTags("Query", queryName)))
await foreach (var val in conn.QueryStreamAsync<T>(query.Sql, query.NamedBindings))
yield return val;
}
// the procedures (message_context and proxy_members, as of writing) have their own metrics tracking elsewhere
// still, including them here for consistency
public async Task<T> QuerySingleProcedure<T>(string queryName, object param)
{
using var conn = await Obtain();
return await conn.QueryFirstAsync<T>(queryName, param, commandType: CommandType.StoredProcedure);
}
public async Task<IEnumerable<T>> QueryProcedure<T>(string queryName, object param)
{
using var conn = await Obtain();
return await conn.QueryAsync<T>(queryName, param, commandType: CommandType.StoredProcedure);
}
}
}

15
PluralKit.Core/Database/IDatabase.cs
View File

@@ -1,10 +1,25 @@
using System;
using System.Collections.Generic;
using System.Runtime.CompilerServices;
using System.Threading.Tasks;
using SqlKata;
namespace PluralKit.Core
{
public interface IDatabase
{
Task ApplyMigrations();
Task<IPKConnection> Obtain();
Task Execute(Func<IPKConnection, Task> func);
Task<T> Execute<T>(Func<IPKConnection, Task<T>> func);
IAsyncEnumerable<T> Execute<T>(Func<IPKConnection, IAsyncEnumerable<T>> func);
Task<int> ExecuteQuery(Query q, string extraSql = "", [CallerMemberName] string queryName = "");
Task<T> QueryFirst<T>(Query q, string extraSql = "", [CallerMemberName] string queryName = "");
Task<T> QueryFirst<T>(IPKConnection? conn, Query q, string extraSql = "", [CallerMemberName] string queryName = "");
Task<IEnumerable<T>> Query<T>(Query q, [CallerMemberName] string queryName = "");
IAsyncEnumerable<T> QueryStream<T>(Query q, [CallerMemberName] string queryName = "");
Task<T> QuerySingleProcedure<T>(string queryName, object param);
Task<IEnumerable<T>> QueryProcedure<T>(string queryName, object param);
}
}

13
PluralKit.Core/Database/Repository/ModelRepository.Account.cs
View File

@@ -1,21 +1,16 @@
using System.Collections.Generic;
using System.Data;
using System.Threading.Tasks;
using Dapper;
using SqlKata;
namespace PluralKit.Core
{
public partial class ModelRepository
{
public async Task UpdateAccount(IPKConnection conn, ulong id, AccountPatch patch)
public async Task UpdateAccount(ulong id, AccountPatch patch)
{
_logger.Information("Updated account {accountId}: {@AccountPatch}", id, patch);
var (query, pms) = patch.Apply(UpdateQueryBuilder.Update("accounts", "uid = @uid"))
.WithConstant("uid", id)
.Build();
await conn.ExecuteAsync(query, pms);
var query = patch.Apply(new Query("accounts").Where("uid", id));
await _db.ExecuteQuery(query);
}
}
}

31
PluralKit.Core/Database/Repository/ModelRepository.CommandMessage.cs
View File

@@ -1,22 +1,33 @@
using System.Threading.Tasks;
using Dapper;
using SqlKata;
namespace PluralKit.Core
{
public partial class ModelRepository
{
public Task SaveCommandMessage(IPKConnection conn, ulong messageId, ulong channelId, ulong authorId) =>
conn.QueryAsync("insert into command_messages (message_id, channel_id, author_id) values (@Message, @Channel, @Author)",
new { Message = messageId, Channel = channelId, Author = authorId });
public Task SaveCommandMessage(ulong messageId, ulong channelId, ulong authorId)
{
var query = new Query("command_messages").AsInsert(new
{
message_id = messageId,
channel_id = channelId,
author_id = authorId,
});
return _db.ExecuteQuery(query);
}
public Task<CommandMessage?> GetCommandMessage(IPKConnection conn, ulong messageId) =>
conn.QuerySingleOrDefaultAsync<CommandMessage>("select * from command_messages where message_id = @Message",
new { Message = messageId });
public Task<CommandMessage?> GetCommandMessage(ulong messageId)
{
var query = new Query("command_messages").Where("message_id", messageId);
return _db.QueryFirst<CommandMessage?>(query);
}
public Task<int> DeleteCommandMessagesBefore(IPKConnection conn, ulong messageIdThreshold) =>
conn.ExecuteAsync("delete from command_messages where message_id < @Threshold",
new { Threshold = messageIdThreshold });
public Task<int> DeleteCommandMessagesBefore(ulong messageIdThreshold)
{
var query = new Query("command_messages").AsDelete().Where("message_id", "<", messageIdThreshold);
return _db.QueryFirst<int>(query);
}
}
public class CommandMessage

28
PluralKit.Core/Database/Repository/ModelRepository.Context.cs
View File

@@ -1,25 +1,23 @@
using System.Collections.Generic;
using System.Data;
using System.Threading.Tasks;
using Dapper;
namespace PluralKit.Core
{
public partial class ModelRepository
{
public Task<MessageContext> GetMessageContext(IPKConnection conn, ulong account, ulong guild, ulong channel)
{
return conn.QueryFirstAsync<MessageContext>("message_context",
new { account_id = account, guild_id = guild, channel_id = channel },
commandType: CommandType.StoredProcedure);
}
public Task<MessageContext> GetMessageContext(ulong account, ulong guild, ulong channel)
=> _db.QuerySingleProcedure<MessageContext>("message_context", new
{
account_id = account,
guild_id = guild,
channel_id = channel
});
public Task<IEnumerable<ProxyMember>> GetProxyMembers(IPKConnection conn, ulong account, ulong guild)
{
return conn.QueryAsync<ProxyMember>("proxy_members",
new { account_id = account, guild_id = guild },
commandType: CommandType.StoredProcedure);
}
public Task<IEnumerable<ProxyMember>> GetProxyMembers(ulong account, ulong guild)
=> _db.QueryProcedure<ProxyMember>("proxy_members", new
{
account_id = account,
guild_id = guild
});
}
}

114
PluralKit.Core/Database/Repository/ModelRepository.Group.cs
View File

@@ -1,89 +1,77 @@
#nullable enable
using System;
using System.Collections.Generic;
using System.Data;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Dapper;
using SqlKata;
namespace PluralKit.Core
{
public partial class ModelRepository
{
public Task<PKGroup?> GetGroupByName(IPKConnection conn, SystemId system, string name) =>
conn.QueryFirstOrDefaultAsync<PKGroup?>("select * from groups where system = @System and lower(Name) = lower(@Name)", new { System = system, Name = name });
public Task<PKGroup?> GetGroupByDisplayName(IPKConnection conn, SystemId system, string display_name) =>
conn.QueryFirstOrDefaultAsync<PKGroup?>("select * from groups where system = @System and lower(display_name) = lower(@Name)", new { System = system, Name = display_name });
public Task<PKGroup?> GetGroupByHid(IPKConnection conn, string hid, SystemId? system = null)
=> conn.QueryFirstOrDefaultAsync<PKGroup?>(
"select * from groups where hid = @hid" + (system != null ? " and system = @System" : ""),
new { hid = hid.ToLowerInvariant(), System = system }
);
public Task<PKGroup?> GetGroupByGuid(IPKConnection conn, Guid guid) =>
conn.QueryFirstOrDefaultAsync<PKGroup?>("select * from groups where uuid = @Uuid", new { Uuid = guid });
public Task<int> GetGroupMemberCount(IPKConnection conn, GroupId id, PrivacyLevel? privacyFilter = null)
public Task<PKGroup?> GetGroupByName(SystemId system, string name)
{
var query = new StringBuilder("select count(*) from group_members");
if (privacyFilter != null)
query.Append(" inner join members on group_members.member_id = members.id");
query.Append(" where group_members.group_id = @Id");
if (privacyFilter != null)
query.Append(" and members.member_visibility = @PrivacyFilter");
return conn.QuerySingleOrDefaultAsync<int>(query.ToString(), new { Id = id, PrivacyFilter = privacyFilter });
var query = new Query("groups").Where("system", system).WhereRaw("lower(name) = lower(?)", name.ToLower());
return _db.QueryFirst<PKGroup?>(query);
}
public async Task<PKGroup> CreateGroup(IPKConnection conn, SystemId system, string name, IDbTransaction? transaction = null)
public Task<PKGroup?> GetGroupByDisplayName(SystemId system, string display_name)
{
var group = await conn.QueryFirstAsync<PKGroup>(
"insert into groups (hid, system, name) values (find_free_group_hid(), @System, @Name) returning *",
new { System = system, Name = name }, transaction);
var query = new Query("groups").Where("system", system).WhereRaw("lower(display_name) = lower(?)", display_name.ToLower());
return _db.QueryFirst<PKGroup?>(query);
}
public Task<PKGroup?> GetGroupByHid(string hid, SystemId? system = null)
{
var query = new Query("groups").Where("hid", hid.ToLower());
if (system != null)
query = query.Where("system", system);
return _db.QueryFirst<PKGroup?>(query);
}
public Task<PKGroup?> GetGroupByGuid(Guid uuid)
{
var query = new Query("groups").Where("uuid", uuid);
return _db.QueryFirst<PKGroup?>(query);
}
public Task<int> GetGroupMemberCount(GroupId id, PrivacyLevel? privacyFilter = null)
{
var query = new Query("group_members")
.SelectRaw("count(*)")
.Where("group_members.group_id", id);
if (privacyFilter != null) query = query
.Join("members", "group_members.member_id", "members.id")
.Where("members.member_visibility", privacyFilter);
return _db.QueryFirst<int>(query);
}
public async Task<PKGroup> CreateGroup(SystemId system, string name, IPKConnection? conn = null)
{
var query = new Query("groups").AsInsert(new
{
hid = new UnsafeLiteral("find_free_group_hid()"),
system = system,
name = name
});
var group = await _db.QueryFirst<PKGroup>(conn, query, extraSql: "returning *");
_logger.Information("Created group {GroupId} in system {SystemId}: {GroupName}", group.Id, system, name);
return group;
}
public Task<PKGroup> UpdateGroup(IPKConnection conn, GroupId id, GroupPatch patch, IDbTransaction? transaction = null)
public Task<PKGroup> UpdateGroup(GroupId id, GroupPatch patch, IPKConnection? conn = null)
{
_logger.Information("Updated {GroupId}: {@GroupPatch}", id, patch);
var (query, pms) = patch.Apply(UpdateQueryBuilder.Update("groups", "id = @id"))
.WithConstant("id", id)
.Build("returning *");
return conn.QueryFirstAsync<PKGroup>(query, pms, transaction);
var query = patch.Apply(new Query("groups").Where("id", id));
return _db.QueryFirst<PKGroup>(conn, query, extraSql: "returning *");
}
public Task DeleteGroup(IPKConnection conn, GroupId group)
public Task DeleteGroup(GroupId group)
{
_logger.Information("Deleted {GroupId}", group);
return conn.ExecuteAsync("delete from groups where id = @Id", new { Id = @group });
}
public async Task AddMembersToGroup(IPKConnection conn, GroupId group,
IReadOnlyCollection<MemberId> members)
{
await using var w =
conn.BeginBinaryImport("copy group_members (group_id, member_id) from stdin (format binary)");
foreach (var member in members)
{
await w.StartRowAsync();
await w.WriteAsync(group.Value);
await w.WriteAsync(member.Value);
}
await w.CompleteAsync();
_logger.Information("Added members to {GroupId}: {MemberIds}", group, members);
}
public Task RemoveMembersFromGroup(IPKConnection conn, GroupId group,
IReadOnlyCollection<MemberId> members)
{
_logger.Information("Removed members from {GroupId}: {MemberIds}", group, members);
return conn.ExecuteAsync("delete from group_members where group_id = @Group and member_id = any(@Members)",
new { Group = @group, Members = members.ToArray() });
var query = new Query("groups").AsDelete().Where("id", group);
return _db.ExecuteQuery(query);
}
}
}

55
PluralKit.Core/Database/Repository/ModelRepository.GroupMember.cs
View File

@@ -1,21 +1,25 @@
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Dapper;
using SqlKata;
namespace PluralKit.Core
{
public partial class ModelRepository
{
public IAsyncEnumerable<PKGroup> GetMemberGroups(IPKConnection conn, MemberId id) =>
conn.QueryStreamAsync<PKGroup>(
"select groups.* from group_members inner join groups on group_members.group_id = groups.id where group_members.member_id = @Id",
new { Id = id });
public async Task AddGroupsToMember(IPKConnection conn, MemberId member, IReadOnlyCollection<GroupId> groups)
public IAsyncEnumerable<PKGroup> GetMemberGroups(MemberId id)
{
var query = new Query("group_members")
.Select("groups.*")
.Join("groups", "group_members.group_id", "groups.id")
.Where("group_members.member_id", id);
return _db.QueryStream<PKGroup>(query);
}
// todo: add this to metrics tracking
public async Task AddGroupsToMember(MemberId member, IReadOnlyCollection<GroupId> groups)
{
await using var conn = await _db.Obtain();
await using var w =
conn.BeginBinaryImport("copy group_members (group_id, member_id) from stdin (format binary)");
foreach (var group in groups)
@@ -29,12 +33,39 @@ namespace PluralKit.Core
_logger.Information("Added member {MemberId} to groups {GroupIds}", member, groups);
}
public Task RemoveGroupsFromMember(IPKConnection conn, MemberId member, IReadOnlyCollection<GroupId> groups)
public Task RemoveGroupsFromMember(MemberId member, IReadOnlyCollection<GroupId> groups)
{
_logger.Information("Removed groups from {MemberId}: {GroupIds}", member, groups);
return conn.ExecuteAsync("delete from group_members where member_id = @Member and group_id = any(@Groups)",
new { Member = @member, Groups = groups.ToArray() });
var query = new Query("group_members").AsDelete()
.Where("member_id", member)
.WhereIn("group_id", groups);
return _db.ExecuteQuery(query);
}
// todo: add this to metrics tracking
public async Task AddMembersToGroup(GroupId group, IReadOnlyCollection<MemberId> members)
{
await using var conn = await _db.Obtain();
await using var w =
conn.BeginBinaryImport("copy group_members (group_id, member_id) from stdin (format binary)");
foreach (var member in members)
{
await w.StartRowAsync();
await w.WriteAsync(group.Value);
await w.WriteAsync(member.Value);
}
await w.CompleteAsync();
_logger.Information("Added members to {GroupId}: {MemberIds}", group, members);
}
public Task RemoveMembersFromGroup(GroupId group, IReadOnlyCollection<MemberId> members)
{
_logger.Information("Removed members from {GroupId}: {MemberIds}", group, members);
var query = new Query("group_members").AsDelete()
.Where("group_id", group)
.WhereIn("member_id", members);
return _db.ExecuteQuery(query);
}
}
}

76
PluralKit.Core/Database/Repository/ModelRepository.Guild.cs
View File

@@ -1,53 +1,63 @@
using System.Threading.Tasks;
using Dapper;
using SqlKata;
namespace PluralKit.Core
{
public partial class ModelRepository
{
public Task UpsertGuild(IPKConnection conn, ulong guild, GuildPatch patch)
public Task<GuildConfig> GetGuild(ulong guild)
{
var query = new Query("servers").AsInsert(new { id = guild });
// sqlkata doesn't support postgres on conflict, so we just hack it on here
return _db.QueryFirst<GuildConfig>(query, "on conflict (id) do update set id = @$1 returning *");
}
public Task UpdateGuild(ulong guild, GuildPatch patch)
{
_logger.Information("Updated guild {GuildId}: {@GuildPatch}", guild, patch);
var (query, pms) = patch.Apply(UpdateQueryBuilder.Upsert("servers", "id"))
.WithConstant("id", guild)
.Build();
return conn.ExecuteAsync(query, pms);
var query = patch.Apply(new Query("servers").Where("id", guild));
return _db.ExecuteQuery(query);
}
public Task UpsertSystemGuild(IPKConnection conn, SystemId system, ulong guild,
SystemGuildPatch patch)
public Task<SystemGuildSettings> GetSystemGuild(ulong guild, SystemId system)
{
var query = new Query("system_guild").AsInsert(new
{
guild = guild,
system = system
});
return _db.QueryFirst<SystemGuildSettings>(query,
extraSql: "on conflict (guild, system) do update set guild = $1, system = $2 returning *"
);
}
public Task UpdateSystemGuild(SystemId system, ulong guild, SystemGuildPatch patch)
{
_logger.Information("Updated {SystemId} in guild {GuildId}: {@SystemGuildPatch}", system, guild, patch);
var (query, pms) = patch.Apply(UpdateQueryBuilder.Upsert("system_guild", "system, guild"))
.WithConstant("system", system)
.WithConstant("guild", guild)
.Build();
return conn.ExecuteAsync(query, pms);
var query = patch.Apply(new Query("system_guild").Where("system", system).Where("guild", guild));
return _db.ExecuteQuery(query);
}
public Task UpsertMemberGuild(IPKConnection conn, MemberId member, ulong guild,
MemberGuildPatch patch)
public Task<MemberGuildSettings> GetMemberGuild(ulong guild, MemberId member)
{
var query = new Query("member_guild").AsInsert(new
{
guild = guild,
member = member
});
return _db.QueryFirst<MemberGuildSettings>(query,
extraSql: "on conflict (guild, member) do update set guild = $1, member = $2 returning *"
);
}
public Task UpdateMemberGuild(MemberId member, ulong guild, MemberGuildPatch patch)
{
_logger.Information("Updated {MemberId} in guild {GuildId}: {@MemberGuildPatch}", member, guild, patch);
var (query, pms) = patch.Apply(UpdateQueryBuilder.Upsert("member_guild", "member, guild"))
.WithConstant("member", member)
.WithConstant("guild", guild)
.Build();
return conn.ExecuteAsync(query, pms);
var query = patch.Apply(new Query("member_guild").Where("member", member).Where("guild", guild));
return _db.ExecuteQuery(query);
}
public Task<GuildConfig> GetGuild(IPKConnection conn, ulong guild) =>
conn.QueryFirstAsync<GuildConfig>("insert into servers (id) values (@guild) on conflict (id) do update set id = @guild returning *", new { guild });
public Task<SystemGuildSettings> GetSystemGuild(IPKConnection conn, ulong guild, SystemId system) =>
conn.QueryFirstAsync<SystemGuildSettings>(
"insert into system_guild (guild, system) values (@guild, @system) on conflict (guild, system) do update set guild = @guild, system = @system returning *",
new { guild, system });
public Task<MemberGuildSettings> GetMemberGuild(IPKConnection conn, ulong guild, MemberId member) =>
conn.QueryFirstAsync<MemberGuildSettings>(
"insert into member_guild (guild, member) values (@guild, @member) on conflict (guild, member) do update set guild = @guild, member = @member returning *",
new { guild, member });
}
}

86
PluralKit.Core/Database/Repository/ModelRepository.Member.cs
View File

@@ -1,55 +1,77 @@
#nullable enable
using System;
using System.Data;
using System.Threading.Tasks;
using Dapper;
using SqlKata;
namespace PluralKit.Core
{
public partial class ModelRepository
{
public Task<PKMember?> GetMember(IPKConnection conn, MemberId id) =>
conn.QueryFirstOrDefaultAsync<PKMember?>("select * from members where id = @id", new { id });
public Task<PKMember?> GetMemberByHid(IPKConnection conn, string hid, SystemId? system = null)
=> conn.QuerySingleOrDefaultAsync<PKMember?>(
"select * from members where hid = @Hid" + (system != null ? " and system = @System" : ""),
new { Hid = hid.ToLower(), System = system }
);
public Task<PKMember?> GetMemberByGuid(IPKConnection conn, Guid guid) =>
conn.QuerySingleOrDefaultAsync<PKMember?>("select * from members where uuid = @Uuid", new { Uuid = guid });
public Task<PKMember?> GetMemberByName(IPKConnection conn, SystemId system, string name) =>
conn.QueryFirstOrDefaultAsync<PKMember?>("select * from members where lower(name) = lower(@Name) and system = @SystemID", new { Name = name, SystemID = system });
public Task<PKMember?> GetMemberByDisplayName(IPKConnection conn, SystemId system, string name) =>
conn.QueryFirstOrDefaultAsync<PKMember?>("select * from members where lower(display_name) = lower(@Name) and system = @SystemID", new { Name = name, SystemID = system });
public async Task<PKMember> CreateMember(IPKConnection conn, SystemId id, string memberName, IDbTransaction? transaction = null)
public Task<PKMember?> GetMember(MemberId id)
{
var member = await conn.QueryFirstAsync<PKMember>(
"insert into members (hid, system, name) values (find_free_member_hid(), @SystemId, @Name) returning *",
new { SystemId = id, Name = memberName }, transaction);
var query = new Query("members").Where("id", id);
return _db.QueryFirst<PKMember?>(query);
}
public Task<PKMember?> GetMemberByHid(string hid, SystemId? system = null)
{
var query = new Query("members").Where("hid", hid.ToLower());
if (system != null)
query = query.Where("system", system);
return _db.QueryFirst<PKMember?>(query);
}
public Task<PKMember?> GetMemberByGuid(Guid uuid)
{
var query = new Query("members").Where("uuid", uuid);
return _db.QueryFirst<PKMember?>(query);
}
public Task<PKMember?> GetMemberByName(SystemId system, string name)
{
var query = new Query("members").WhereRaw(
"lower(name) = lower(?)",
name.ToLower()
).Where("system", system);
return _db.QueryFirst<PKMember?>(query);
}
public Task<PKMember?> GetMemberByDisplayName(SystemId system, string name)
{
var query = new Query("members").WhereRaw(
"lower(display_name) = lower(?)",
name.ToLower()
).Where("system", system);
return _db.QueryFirst<PKMember?>(query);
}
public async Task<PKMember> CreateMember(SystemId systemId, string memberName, IPKConnection? conn = null)
{
var query = new Query("members").AsInsert(new
{
hid = new UnsafeLiteral("find_free_member_hid()"),
system = systemId,
name = memberName
});
var member = await _db.QueryFirst<PKMember>(conn, query, "returning *");
_logger.Information("Created {MemberId} in {SystemId}: {MemberName}",
member.Id, id, memberName);
member.Id, systemId, memberName);
return member;
}
public Task<PKMember> UpdateMember(IPKConnection conn, MemberId id, MemberPatch patch, IDbTransaction? transaction = null)
public Task<PKMember> UpdateMember(MemberId id, MemberPatch patch, IPKConnection? conn = null)
{
_logger.Information("Updated {MemberId}: {@MemberPatch}", id, patch);
var (query, pms) = patch.Apply(UpdateQueryBuilder.Update("members", "id = @id"))
.WithConstant("id", id)
.Build("returning *");
return conn.QueryFirstAsync<PKMember>(query, pms, transaction);
var query = patch.Apply(new Query("members").Where("id", id));
return _db.QueryFirst<PKMember>(conn, query);
}
public Task DeleteMember(IPKConnection conn, MemberId id)
public Task DeleteMember(MemberId id)
{
_logger.Information("Deleted {MemberId}", id);
return conn.ExecuteAsync("delete from members where id = @Id", new { Id = id });
var query = new Query("members").AsDelete().Where("id", id);
return _db.ExecuteQuery(query);
}
}
}

47
PluralKit.Core/Database/Repository/ModelRepository.Message.cs
View File

@@ -4,17 +4,30 @@ using System.Threading.Tasks;
using Dapper;
using SqlKata;
namespace PluralKit.Core
{
public partial class ModelRepository
{
public async Task AddMessage(IPKConnection conn, PKMessage msg)
public Task AddMessage(PKMessage msg)
{
// "on conflict do nothing" in the (pretty rare) case of duplicate events coming in from Discord, which would lead to a DB error before
await conn.ExecuteAsync("insert into messages(mid, guild, channel, member, sender, original_mid) values(@Mid, @Guild, @Channel, @Member, @Sender, @OriginalMid) on conflict do nothing", msg);
var query = new Query("messages").AsInsert(new
{
mid = msg.Mid,
guild = msg.Guild,
channel = msg.Channel,
member = msg.Member,
sender = msg.Sender,
original_mid = msg.OriginalMid,
});
_logger.Debug("Stored message {@StoredMessage} in channel {Channel}", msg, msg.Channel);
// "on conflict do nothing" in the (pretty rare) case of duplicate events coming in from Discord, which would lead to a DB error before
return _db.ExecuteQuery(query, extraSql: "on conflict do nothing");
}
// todo: add a Mapper to QuerySingle and move this to SqlKata
public async Task<FullMessage?> GetMessage(IPKConnection conn, ulong id)
{
FullMessage Mapper(PKMessage msg, PKMember member, PKSystem system) =>
@@ -26,34 +39,36 @@ namespace PluralKit.Core
return result.FirstOrDefault();
}
public async Task DeleteMessage(IPKConnection conn, ulong id)
public async Task DeleteMessage(ulong id)
{
var rowCount = await conn.ExecuteAsync("delete from messages where mid = @Id", new { Id = id });
var query = new Query("messages").AsDelete().Where("mid", id);
var rowCount = await _db.ExecuteQuery(query);
if (rowCount > 0)
_logger.Information("Deleted message {MessageId} from database", id);
}
public async Task DeleteMessagesBulk(IPKConnection conn, IReadOnlyCollection<ulong> ids)
public async Task DeleteMessagesBulk(IReadOnlyCollection<ulong> ids)
{
// Npgsql doesn't support ulongs in general - we hacked around it for plain ulongs but tbh not worth it for collections of ulong
// Hence we map them to single longs, which *are* supported (this is ok since they're Technically (tm) stored as signed longs in the db anyway)
var rowCount = await conn.ExecuteAsync("delete from messages where mid = any(@Ids)",
new { Ids = ids.Select(id => (long)id).ToArray() });
var query = new Query("messages").AsDelete().WhereIn("mid", ids.Select(id => (long)id).ToArray());
var rowCount = await _db.ExecuteQuery(query);
if (rowCount > 0)
_logger.Information("Bulk deleted messages ({FoundCount} found) from database: {MessageIds}", rowCount,
ids);
}
public async Task<PKMessage?> GetLastMessage(IPKConnection conn, ulong guildId, ulong channelId, ulong accountId)
public Task<PKMessage?> GetLastMessage(ulong guildId, ulong channelId, ulong accountId)
{
// Want to index scan on the (guild, sender, mid) index so need the additional constraint
return await conn.QuerySingleOrDefaultAsync<PKMessage>(
"select * from messages where guild = @Guild and channel = @Channel and sender = @Sender order by mid desc limit 1", new
{
Guild = guildId,
Channel = channelId,
Sender = accountId
});
var query = new Query("messages")
.Where("guild", guildId)
.Where("channel", channelId)
.Where("sender", accountId)
.OrderByDesc("mid")
.Limit(1);
return _db.QueryFirst<PKMessage?>(query);
}
}

49
PluralKit.Core/Database/Repository/ModelRepository.Switch.cs
View File

@@ -9,8 +9,11 @@ using NodaTime;
using NpgsqlTypes;
using SqlKata;
namespace PluralKit.Core
{
// todo: move the rest of the queries in here to SqlKata, if possible
public partial class ModelRepository
{
public async Task<PKSwitch> AddSwitch(IPKConnection conn, SystemId system, IReadOnlyCollection<MemberId> members)
@@ -69,40 +72,44 @@ namespace PluralKit.Core
_logger.Information("Updated {SwitchId} members: {Members}", switchId, members);
}
public async Task MoveSwitch(IPKConnection conn, SwitchId id, Instant time)
public Task MoveSwitch(SwitchId id, Instant time)
{
await conn.ExecuteAsync("update switches set timestamp = @Time where id = @Id",
new { Time = time, Id = id });
_logger.Information("Updated {SwitchId} timestamp: {SwitchTimestamp}", id, time);
var query = new Query("switches").AsUpdate(new { timestamp = time }).Where("id", id);
return _db.ExecuteQuery(query);
}
public async Task DeleteSwitch(IPKConnection conn, SwitchId id)
public Task DeleteSwitch(SwitchId id)
{
await conn.ExecuteAsync("delete from switches where id = @Id", new { Id = id });
_logger.Information("Deleted {Switch}", id);
var query = new Query("switches").AsDelete().Where("id", id);
return _db.ExecuteQuery(query);
}
public async Task DeleteAllSwitches(IPKConnection conn, SystemId system)
public Task DeleteAllSwitches(SystemId system)
{
await conn.ExecuteAsync("delete from switches where system = @Id", new { Id = system });
_logger.Information("Deleted all switches in {SystemId}", system);
var query = new Query("switches").AsDelete().Where("system", system);
return _db.ExecuteQuery(query);
}
public IAsyncEnumerable<PKSwitch> GetSwitches(IPKConnection conn, SystemId system)
public IAsyncEnumerable<PKSwitch> GetSwitches(SystemId system)
{
// TODO: refactor the PKSwitch data structure to somehow include a hydrated member list
return conn.QueryStreamAsync<PKSwitch>(
"select * from switches where system = @System order by timestamp desc",
new { System = system });
var query = new Query("switches").Where("system", system).OrderByDesc("timestamp");
return _db.QueryStream<PKSwitch>(query);
}
public Task<PKSwitch> GetSwitchByUuid(IPKConnection conn, Guid uuid) =>
conn.QuerySingleOrDefaultAsync<PKSwitch>("select * from switches where uuid = @Uuid", new { Uuid = uuid });
public async Task<int> GetSwitchCount(IPKConnection conn, SystemId system)
public Task<PKSwitch> GetSwitchByUuid(Guid uuid)
{
return await conn.QuerySingleAsync<int>("select count(*) from switches where system = @Id", new { Id = system });
var query = new Query("switches").Where("uuid", uuid);
return _db.QueryFirst<PKSwitch>(query);
}
public Task<int> GetSwitchCount(SystemId system)
{
var query = new Query("switches").SelectRaw("count(*)").Where("system", system);
return _db.QueryFirst<int>(query);
}
public async IAsyncEnumerable<SwitchMembersListEntry> GetSwitchMembersList(IPKConnection conn,
@@ -149,9 +156,11 @@ namespace PluralKit.Core
new { Switch = sw });
}
public async Task<PKSwitch> GetLatestSwitch(IPKConnection conn, SystemId system) =>
// TODO: should query directly for perf
await GetSwitches(conn, system).FirstOrDefaultAsync();
public Task<PKSwitch> GetLatestSwitch(SystemId system)
{
var query = new Query("switches").Where("system", system).OrderByDesc("timestamp").Limit(1);
return _db.QueryFirst<PKSwitch>(query);
}
public async Task<IEnumerable<SwitchListEntry>> GetPeriodFronters(IPKConnection conn,
SystemId system, GroupId? group, Instant periodStart,

135
PluralKit.Core/Database/Repository/ModelRepository.System.cs
View File

@@ -1,93 +1,120 @@
#nullable enable
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;
using Dapper;
using SqlKata;
namespace PluralKit.Core
{
public partial class ModelRepository
{
public Task<PKSystem?> GetSystem(IPKConnection conn, SystemId id) =>
conn.QueryFirstOrDefaultAsync<PKSystem?>("select * from systems where id = @id", new { id });
public Task<PKSystem?> GetSystemByGuid(IPKConnection conn, Guid id) =>
conn.QueryFirstOrDefaultAsync<PKSystem?>("select * from systems where uuid = @id", new { id });
public Task<PKSystem?> GetSystemByAccount(IPKConnection conn, ulong accountId) =>
conn.QuerySingleOrDefaultAsync<PKSystem?>(
"select systems.* from systems, accounts where accounts.system = systems.id and accounts.uid = @Id",
new { Id = accountId });
public Task<PKSystem?> GetSystemByHid(IPKConnection conn, string hid) =>
conn.QuerySingleOrDefaultAsync<PKSystem?>("select * from systems where systems.hid = @Hid",
new { Hid = hid.ToLower() });
public Task<IEnumerable<ulong>> GetSystemAccounts(IPKConnection conn, SystemId system) =>
conn.QueryAsync<ulong>("select uid from accounts where system = @Id", new { Id = system });
public IAsyncEnumerable<PKMember> GetSystemMembers(IPKConnection conn, SystemId system) =>
conn.QueryStreamAsync<PKMember>("select * from members where system = @SystemID", new { SystemID = system });
public IAsyncEnumerable<PKGroup> GetSystemGroups(IPKConnection conn, SystemId system) =>
conn.QueryStreamAsync<PKGroup>("select * from groups where system = @System", new { System = system });
public Task<int> GetSystemMemberCount(IPKConnection conn, SystemId id, PrivacyLevel? privacyFilter = null)
public Task<PKSystem?> GetSystem(SystemId id)
{
var query = new StringBuilder("select count(*) from members where system = @Id");
if (privacyFilter != null)
query.Append($" and member_visibility = {(int)privacyFilter.Value}");
return conn.QuerySingleAsync<int>(query.ToString(), new { Id = id });
var query = new Query("systems").Where("id", id);
return _db.QueryFirst<PKSystem?>(query);
}
public Task<int> GetSystemGroupCount(IPKConnection conn, SystemId id, PrivacyLevel? privacyFilter = null)
public Task<PKSystem?> GetSystemByGuid(Guid id)
{
var query = new StringBuilder("select count(*) from groups where system = @Id");
if (privacyFilter != null)
query.Append($" and visibility = {(int)privacyFilter.Value}");
return conn.QuerySingleAsync<int>(query.ToString(), new { Id = id });
var query = new Query("systems").Where("uuid", id);
return _db.QueryFirst<PKSystem?>(query);
}
public async Task<PKSystem> CreateSystem(IPKConnection conn, string? systemName = null, IPKTransaction? tx = null)
public Task<PKSystem?> GetSystemByAccount(ulong accountId)
{
var system = await conn.QuerySingleAsync<PKSystem>(
"insert into systems (hid, name) values (find_free_system_hid(), @Name) returning *",
new { Name = systemName },
transaction: tx);
var query = new Query("accounts").Select("systems.*").LeftJoin("systems", "systems.id", "accounts.system", "=").Where("uid", accountId);
return _db.QueryFirst<PKSystem?>(query);
}
public Task<PKSystem?> GetSystemByHid(string hid)
{
var query = new Query("systems").Where("hid", hid.ToLower());
return _db.QueryFirst<PKSystem?>(query);
}
public Task<IEnumerable<ulong>> GetSystemAccounts(SystemId system)
{
var query = new Query("accounts").Select("uid").Where("system", system);
return _db.Query<ulong>(query);
}
public IAsyncEnumerable<PKMember> GetSystemMembers(SystemId system)
{
var query = new Query("members").Where("system", system);
return _db.QueryStream<PKMember>(query);
}
public IAsyncEnumerable<PKGroup> GetSystemGroups(SystemId system)
{
var query = new Query("groups").Where("system", system);
return _db.QueryStream<PKGroup>(query);
}
public Task<int> GetSystemMemberCount(SystemId system, PrivacyLevel? privacyFilter = null)
{
var query = new Query("members").SelectRaw("count(*)").Where("system", system);
if (privacyFilter != null)
query.Where("member_visibility", (int)privacyFilter.Value);
return _db.QueryFirst<int>(query);
}
public Task<int> GetSystemGroupCount(SystemId system, PrivacyLevel? privacyFilter = null)
{
var query = new Query("groups").SelectRaw("count(*)").Where("system", system);
if (privacyFilter != null)
query.Where("visibility", (int)privacyFilter.Value);
return _db.QueryFirst<int>(query);
}
public async Task<PKSystem> CreateSystem(string? systemName = null, IPKConnection? conn = null)
{
var query = new Query("systems").AsInsert(new
{
hid = new UnsafeLiteral("find_free_system_hid()"),
name = systemName
});
var system = await _db.QueryFirst<PKSystem>(conn, query, extraSql: "returning *");
_logger.Information("Created {SystemId}", system.Id);
return system;
}
public Task<PKSystem> UpdateSystem(IPKConnection conn, SystemId id, SystemPatch patch, IPKTransaction? tx = null)
public Task<PKSystem> UpdateSystem(SystemId id, SystemPatch patch, IPKConnection? conn = null)
{
_logger.Information("Updated {SystemId}: {@SystemPatch}", id, patch);
var (query, pms) = patch.Apply(UpdateQueryBuilder.Update("systems", "id = @id"))
.WithConstant("id", id)
.Build("returning *");
return conn.QueryFirstAsync<PKSystem>(query, pms, transaction: tx);
var query = patch.Apply(new Query("systems").Where("id", id));
return _db.QueryFirst<PKSystem>(conn, query, extraSql: "returning *");
}
public async Task AddAccount(IPKConnection conn, SystemId system, ulong accountId)
public Task AddAccount(SystemId system, ulong accountId)
{
// We have "on conflict do nothing" since linking an account when it's already linked to the same system is idempotent
// This is used in import/export, although the pk;link command checks for this case beforehand
await conn.ExecuteAsync("insert into accounts (uid, system) values (@Id, @SystemId) on conflict do nothing",
new { Id = accountId, SystemId = system });
var query = new Query("accounts").AsInsert(new
{
system = system,
uid = accountId,
});
_logger.Information("Linked account {UserId} to {SystemId}", accountId, system);
return _db.ExecuteQuery(query, extraSql: "on conflict do nothing");
}
public async Task RemoveAccount(IPKConnection conn, SystemId system, ulong accountId)
public async Task RemoveAccount(SystemId system, ulong accountId)
{
await conn.ExecuteAsync("delete from accounts where uid = @Id and system = @SystemId",
new { Id = accountId, SystemId = system });
var query = new Query("accounts").AsDelete().Where("uid", accountId).Where("system", system);
await _db.ExecuteQuery(query);
_logger.Information("Unlinked account {UserId} from {SystemId}", accountId, system);
}
public Task DeleteSystem(IPKConnection conn, SystemId id)
public Task DeleteSystem(SystemId id)
{
var query = new Query("systems").AsDelete().Where("id", id);
_logger.Information("Deleted {SystemId}", id);
return conn.ExecuteAsync("delete from systems where id = @Id", new { Id = id });
return _db.ExecuteQuery(query);
}
}
}

5
PluralKit.Core/Database/Repository/ModelRepository.cs
View File

@@ -5,10 +5,11 @@ namespace PluralKit.Core
public partial class ModelRepository
{
private readonly ILogger _logger;
public ModelRepository(ILogger logger)
private readonly IDatabase _db;
public ModelRepository(ILogger logger, IDatabase db)
{
_logger = logger.ForContext<ModelRepository>();
_db = db;
}
}
}

28
PluralKit.Core/Database/Utils/QueryPatchWrapper.cs Normal file
View File

@@ -0,0 +1,28 @@
using System;
using System.Collections.Generic;
using SqlKata;
namespace PluralKit.Core
{
internal class QueryPatchWrapper
{
private Dictionary<string, object> _dict = new();
public QueryPatchWrapper With<T>(string columnName, Partial<T> partialValue)
{
if (partialValue.IsPresent)
_dict.Add(columnName, partialValue);
return this;
}
public Query ToQuery(Query q) => q.AsUpdate(_dict);
}
internal static class SqlKataExtensions
{
internal static Query ApplyPatch(this Query query, Func<QueryPatchWrapper, QueryPatchWrapper> func)
=> func(new QueryPatchWrapper()).ToQuery(query);
}
}