mc/messengerapi.Broker
1
0
Fork 0
Code Pull Requests Actions Packages Releases Activity

Broker implementation updated.

Browse Source
This commit is contained in:
masiton
2025-07-05 08:48:14 +02:00
parent 8735510dfc
commit 91a1f7aa61
22 changed files with 587 additions and 510 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
code/MessengerBroker/Dockerfile
View File

@ -1,29 +0,0 @@
# See https://aka.ms/customizecontainer to learn how to customize your debug container and how Visual Studio uses this Dockerfile to build your images for faster debugging.
# This stage is used when running from VS in fast mode (Default for Debug configuration)
FROM mcr.microsoft.com/dotnet/aspnet:9.0 AS base
USER $APP_UID
WORKDIR /app
EXPOSE 8080
# This stage is used to build the service project
FROM mcr.microsoft.com/dotnet/sdk:9.0 AS build
ARG BUILD_CONFIGURATION=Release
WORKDIR /src
COPY ["MessengerBroker/MessengerBroker.csproj", "MessengerBroker/"]
RUN dotnet restore "./MessengerBroker/MessengerBroker.csproj"
COPY . .
WORKDIR "/src/MessengerBroker"
RUN dotnet build "./MessengerBroker.csproj" -c $BUILD_CONFIGURATION -o /app/build
# This stage is used to publish the service project to be copied to the final stage
FROM build AS publish
ARG BUILD_CONFIGURATION=Release
RUN dotnet publish "./MessengerBroker.csproj" -c $BUILD_CONFIGURATION -o /app/publish /p:UseAppHost=false
# This stage is used in production or when running from VS in regular mode (Default when not using the Debug configuration)
FROM base AS final
WORKDIR /app
COPY --from=publish /app/publish .
ENTRYPOINT ["dotnet", "MessengerBroker.dll"]

22
code/MessengerBroker/Extensions/EntityExtensions.cs
View File

@ -1,22 +0,0 @@
using System.Linq.Expressions;
namespace MessengerBroker
{
public static class EntityExtensions
{
public static IQueryable<Db.Model.User> GetUsers(this IQueryable<Db.Model.User> source, Guid[] guids)
{
return source.Where(x => guids.Any(g => g == x.Id));
}
public static IQueryable<Db.Model.User> GetUsersExcept(this IQueryable<Db.Model.User> source, IEnumerable<Guid> guids)
{
return source.Where(x => !guids.Any(g => g == x.Id));
}
public static bool GetUserExists(this IQueryable<Db.Model.User> source, Guid id)
{
return source.Any(x=> x.Id == id);
}
}
}

27
code/MessengerBroker/Factories/BrokerDbContextFactory.cs Normal file
View File

@ -0,0 +1,27 @@
using MessengerApi.Configuration.Model.Persistence;
using MessengerBroker.Configuration.Model;
using MessengerBroker.Db;
using MessengerBroker.Db.Sql;
namespace MessengerBroker.Factories
{
public class BrokerDbContextFactory
{
private readonly BrokerConfiguration configuration;
public BrokerDbContextFactory(BrokerConfiguration configuration)
{
this.configuration = configuration;
}
public BrokerDbContext CreateDbContext()
{
if(this.configuration.BrokerPersistenceConfiguration.PersistenceType == MessengerApi.Configuration.Enums.PersistenceTypes.Sql)
{
return new BrokerSqlDbContext((this.configuration.BrokerPersistenceConfiguration as SqlPersistenceConfiguration).ConnectionString);
}
throw new NotImplementedException();
}
}
}

2
code/MessengerBroker/GlobalUsings.cs Normal file
View File

@ -0,0 +1,2 @@
global using ILogger = portaloggy.ILogger;
global using portaloggy;

28
code/MessengerBroker/Handlers/AuthHandler.cs
View File

@ -1,28 +0,0 @@
namespace MessengerBroker.Handlers
{
public class AuthHandler
{
private readonly Settings settings;
public AuthHandler(Settings settings)
{
this.settings = settings;
}
public Guid? Auth(HttpContext context)
{
var authHeader = context.Request.Headers["Authorization"].ToString();
if (!string.IsNullOrEmpty(authHeader) && authHeader.StartsWith("Bearer "))
{
var token = authHeader.Substring("Bearer ".Length).Trim();
if (Guid.TryParse(token, out Guid brokerId) && this.settings.Slaves.Any(x => x.BrokerId == brokerId))
{
return brokerId;
}
}
return null;
}
}
}

78
code/MessengerBroker/Handlers/CustomBearerAuthenticationHandler.cs Normal file
View File

@ -0,0 +1,78 @@
using MessengerBroker.Configuration.Model;
using MessengerBroker.Models;
using MessengerBroker.Models.Scoped;
using Microsoft.AspNetCore.Authentication;
using Microsoft.Extensions.Caching.Memory;
using Microsoft.Extensions.Options;
using System.Security.Claims;
using System.Text.Encodings.Web;
namespace MessengerBroker.Handlers
{
/// <summary>
/// Validates our permananet API keys sent over as Bearer tokens.
/// </summary>
public class CustomBearerAuthenticationHandler : AuthenticationHandler<AuthenticationSchemeOptions>
{
private readonly IMemoryCache memoryCache;
private readonly BrokerConfiguration configuration;
public CustomBearerAuthenticationHandler(
IOptionsMonitor<AuthenticationSchemeOptions> options,
ILoggerFactory loggerFactory,
UrlEncoder encoder,
IMemoryCache memoryCache,
BrokerConfiguration configuration)
: base(options, loggerFactory, encoder)
{
this.memoryCache = memoryCache;
this.configuration = configuration;
}
protected override Task<AuthenticateResult> HandleAuthenticateAsync()
{
const string HEADER = "Authorization";
const string PREFIX = "Bearer ";
if (!Request.Headers.TryGetValue(HEADER, out var authHeader) ||
!authHeader.ToString().StartsWith(PREFIX))
{
return Task.FromResult(AuthenticateResult.NoResult());
}
var token = authHeader.ToString().Substring(PREFIX.Length).Trim();
if(memoryCache.TryGetValue(token, out CachedIdentity oldCache))
{
var identity = Context.RequestServices.GetRequiredService<Identity>();
identity.Server = oldCache.Server;
return Task.FromResult(AuthenticateResult.Success(new AuthenticationTicket(oldCache.ClaimsPrincipal, Scheme.Name)));
}
else
{
var brokerId = Guid.Parse(token);
var server = configuration.SlaveServers.SingleOrDefault(x => x.BrokerId == brokerId);
var principal = new ClaimsPrincipal(
new ClaimsIdentity(
new List<Claim>
{
new Claim(ClaimTypes.NameIdentifier, token),
new Claim(ClaimTypes.Name, token)
}, Scheme.Name));
var cache = new CachedIdentity
{
ClaimsPrincipal = principal,
Server = server
};
memoryCache.Set(token, cache, TimeSpan.FromMinutes(5));
var identity = Context.RequestServices.GetRequiredService<Identity>();
identity.Server = server;
return Task.FromResult(AuthenticateResult.Success(new AuthenticationTicket(cache.ClaimsPrincipal, Scheme.Name)));
}
}
}
}

81
code/MessengerBroker/Handlers/DataHandler.cs
View File

@ -1,81 +0,0 @@
using MessengerApi.Db;
using MessengerBroker.Db;
using Microsoft.EntityFrameworkCore;
namespace MessengerBroker.Handlers
{
public class DataHandler
{
private readonly Settings _settings;
public Task<Tuple<MessengerApi.Db.Entities.User[], MessengerApi.Db.Entities.UserRoute[]>> GetLocalUsersAndRoutes()
{
var foreignUserIds = (Guid[])null;
using(var broCtx = new BrokerDbContext(this._settings.MessengerBrokerDbConnectionString))
{
foreignUserIds = broCtx.Users.Select(x => x.Id).ToArray();
}
using (var apiCtx = new MessengerDbContext(this._settings.MessengerApiDbConnectionString))
{
var localUsers = apiCtx.Users
.Where(x => !foreignUserIds.Any(f => f == x.Id))
.ToArray();
var localRoutes = apiCtx.UserRoutes
.Include(x => x.From)
.Include(x => x.To)
.Where(x => localUsers.Any(l => l.Id == x.From.Id) && localUsers.Any(l => l.Id == x.To.Id))
.ToArray();
return Task.FromResult(new Tuple<MessengerApi.Db.Entities.User[], MessengerApi.Db.Entities.UserRoute[]>(
localUsers,
localRoutes));
}
}
private Task<MessengerApi.Db.Entities.User[]> GetForeignUsers(Guid brokerId)
{
var foreignUserIds = (Guid[])null;
using (var broCtx = new BrokerDbContext(this._settings.MessengerBrokerDbConnectionString))
{
foreignUserIds = broCtx.Users.Where(x=>x.BrokerId == brokerId).Select(x => x.Id).ToArray();
}
using (var apiCtx = new MessengerDbContext(this._settings.MessengerApiDbConnectionString))
{
var localUsers = apiCtx.Users
.Where(x => !foreignUserIds.Any(f => f == x.Id))
.ToArray();
return Task.FromResult(localUsers);
}
}
public async Task<MessengerApi.Db.Entities.Message[]> GetMessages(Guid brokerId, DateTime sinceUtc)
{
var userIds = (Guid[])null;
if(brokerId == this._settings.BrokerId)
{
// Our messages.
var users = await this.GetLocalUsersAndRoutes();
userIds = users.Item1.Select(x => x.Id).ToArray();
}
else
{
var users = await this.GetForeignUsers(brokerId);
userIds = users.Select(x => x.Id).ToArray();
}
using (var apiCtx = new MessengerDbContext(this._settings.MessengerApiDbConnectionString))
{
var messages = apiCtx.Messages
.Include(x => x.From).Include(x => x.To)
.Where(x => x.CreatedUtc >= sinceUtc && userIds.Contains(x.From.Id))
.ToArray();
return messages;
}
}
}
}

49
code/MessengerBroker/Handlers/Endpoint/MessagesEndpointHandler.cs Normal file
View File

@ -0,0 +1,49 @@
using MessengerApi.Factories;
using MessengerBroker.Configuration.Model;
using MessengerBroker.Factories;
using MessengerBroker.Model.Http;
using Microsoft.EntityFrameworkCore;
namespace MessengerBroker.Handlers.Endpoint
{
public class MessagesEndpointHandler
{
private readonly BrokerConfiguration configuration;
private readonly BrokerDbContextFactory brokerDbContextFactory;
private readonly DbContextFactory messengerDbContextFactory;
public MessagesEndpointHandler(
BrokerConfiguration configuration,
BrokerDbContextFactory brokerDbContextFactory,
DbContextFactory messengerDbContextFactory)
{
this.configuration = configuration;
this.brokerDbContextFactory = brokerDbContextFactory;
this.messengerDbContextFactory = messengerDbContextFactory;
}
public async Task<Messages.MessagesResponse> GetMessages(Messages.MessagesRequest request)
{
using var broCtx = this.brokerDbContextFactory.CreateDbContext();
var brokerMessageIdCollection = broCtx.Messages
.Where(x => x.BrokerId == request.OwnerBrokerId)
.Select(x => x.Id)
.Take(1000)
.ToArray();
using var apiCtx = this.messengerDbContextFactory.CreateDbContext();
var messages = apiCtx.Messages
.Where(x => x.CreatedUtc >= request.SinceUtc && brokerMessageIdCollection.Contains(x.Id))
.ToArray();
var response = new Messages.MessagesResponse
{
Messages = messages
};
return response;
}
}
}

65
code/MessengerBroker/Handlers/Endpoint/UsersEndpointHandler.cs Normal file
View File

@ -0,0 +1,65 @@
using MessengerApi.Factories;
using MessengerBroker.Configuration.Model;
using MessengerBroker.Factories;
using MessengerBroker.Model.Http;
using Microsoft.EntityFrameworkCore;
namespace MessengerBroker.Handlers.Endpoint
{
public class UsersEndpointHandler
{
private readonly BrokerConfiguration configuration;
private readonly BrokerDbContextFactory brokerDbContextFactory;
private readonly DbContextFactory messengerDbContextFactory;
public UsersEndpointHandler(
BrokerConfiguration configuration,
BrokerDbContextFactory brokerDbContextFactory,
DbContextFactory messengerDbContextFactory)
{
this.configuration = configuration;
this.brokerDbContextFactory = brokerDbContextFactory;
this.messengerDbContextFactory = messengerDbContextFactory;
}
public Task<Users.UsersResponse> GetUsers()
{
var foreignUserIds = (Guid[])null;
using(var broCtx = this.brokerDbContextFactory.CreateDbContext())
{
foreignUserIds = broCtx.Users.Select(x => x.Id).ToArray();
}
using (var apiCtx = this.messengerDbContextFactory.CreateDbContext())
{
var localUsers = apiCtx.Users
.Where(x => !foreignUserIds.Any(f => f == x.Id))
.ToArray();
var localRoutes = apiCtx.UserRoutes
.Include(x => x.From)
.Include(x => x.To)
.Where(x => localUsers.Any(l => l.Id == x.From.Id) && localUsers.Any(l => l.Id == x.To.Id))
.ToArray();
return Task.FromResult(new Users.UsersResponse
{
Users = localUsers.Select(x => new Users.UsersResponse.User
{
Id = x.Id,
ApiKey = x.ApiKey,
IsEnabled = x.IsEnabled,
Name = x.Name
}).ToArray(),
UserRoutes = localRoutes.Select(x => new Users.UsersResponse.UserRoute
{
Id = x.Id,
FromId = x.From.Id,
ToId = x.To.Id
}).ToArray()
});
}
}
}
}

328
code/MessengerBroker/Handlers/MasterHandler.cs
View File

@ -1,196 +1,196 @@
using MessengerApi.Db;
using MessengerBroker.Db;
using Microsoft.EntityFrameworkCore;
using System.Text.Json;
//using MessengerApi.Db;
//using MessengerBroker.Db;
//using Microsoft.EntityFrameworkCore;
//using System.Text.Json;
namespace MessengerBroker.Handlers
{
public class MasterHandler
{
private readonly Settings settings;
//namespace MessengerBroker.Handlers
//{
// public class MasterHandler
// {
// private readonly Settings settings;
private readonly HttpClient httpClient = new HttpClient();
// private readonly HttpClient httpClient = new HttpClient();
public async Task BeginSyncingWithMaster(Settings.MasterServer masterServer, CancellationToken ct = default)
{
while (!ct.IsCancellationRequested)
{
await Task.Delay(TimeSpan.FromSeconds(1));
// public async Task BeginSyncingWithMaster(Settings.MasterServer masterServer, CancellationToken ct = default)
// {
// while (!ct.IsCancellationRequested)
// {
// await Task.Delay(TimeSpan.FromSeconds(1));
var usersRequest = new HttpRequestMessage(HttpMethod.Get, $"{masterServer.BrokerApiUrl}/users");
usersRequest.Headers.Authorization = new System.Net.Http.Headers.AuthenticationHeaderValue("Bearer", this.settings.BrokerId.ToString());
// var usersRequest = new HttpRequestMessage(HttpMethod.Get, $"{masterServer.BrokerApiUrl}/users");
// usersRequest.Headers.Authorization = new System.Net.Http.Headers.AuthenticationHeaderValue("Bearer", this.settings.BrokerId.ToString());
var usersResponseMessage = await this.httpClient.SendAsync(usersRequest, ct);
// var usersResponseMessage = await this.httpClient.SendAsync(usersRequest, ct);
if (!usersResponseMessage.IsSuccessStatusCode)
{
continue;
}
// if (!usersResponseMessage.IsSuccessStatusCode)
// {
// continue;
// }
var usersResponse = JsonSerializer.Deserialize<Model.Http.Users.UsersResponse>(await usersResponseMessage.Content.ReadAsStringAsync());
// var usersResponse = JsonSerializer.Deserialize<Model.Http.Users.UsersResponse>(await usersResponseMessage.Content.ReadAsStringAsync());
using (var broCtx = new BrokerDbContext(this.settings.MessengerBrokerDbConnectionString))
using (var apiCtx = new MessengerDbContext(this.settings.MessengerApiDbConnectionString))
{
var updatedUsers = usersResponse.Users.Where(x => broCtx.Users.GetUserExists(x.Id)).ToList();
updatedUsers.ForEach(async x => await this.UpdateUser(x, apiCtx));
// using (var broCtx = new BrokerDbContext(this.settings.MessengerBrokerDbConnectionString))
// using (var apiCtx = new MessengerDbContext(this.settings.MessengerApiDbConnectionString))
// {
// var updatedUsers = usersResponse.Users.Where(x => broCtx.Users.GetUserExists(x.Id)).ToList();
// updatedUsers.ForEach(async x => await this.UpdateUser(x, apiCtx));
var deletedUsers = broCtx.Users.GetUsersExcept(usersResponse.Users.Select(x => x.Id)).ToList();
deletedUsers.ForEach(async x => await this.RemoveUser(x.Id, broCtx, apiCtx));
// var deletedUsers = broCtx.Users.GetUsersExcept(usersResponse.Users.Select(x => x.Id)).ToList();
// deletedUsers.ForEach(async x => await this.RemoveUser(x.Id, broCtx, apiCtx));
var addedUsers = usersResponse.Users.Where(x => !broCtx.Users.GetUserExists(x.Id)).ToList();
addedUsers.ForEach(async x => await this.AddUser(x, masterServer.BrokerId, broCtx, apiCtx));
// var addedUsers = usersResponse.Users.Where(x => !broCtx.Users.GetUserExists(x.Id)).ToList();
// addedUsers.ForEach(async x => await this.AddUser(x, masterServer.BrokerId, broCtx, apiCtx));
foreach (var route in usersResponse.UserRoutes
.Where(r => apiCtx.UserRoutes
.Include(x => x.From)
.Include(x => x.To)
.Any(apir => apir.Id == r.Id && (apir.From.Id != r.FromId || apir.To.Id != r.ToId))))
{
var existing = apiCtx.UserRoutes.Include(x => x.From).Include(x => x.To).Single(x => x.Id == route.Id);
existing.From = apiCtx.Users.Single(x => x.Id == route.FromId);
existing.To = apiCtx.Users.Single(x => x.Id == route.ToId);
}
// foreach (var route in usersResponse.UserRoutes
// .Where(r => apiCtx.UserRoutes
// .Include(x => x.From)
// .Include(x => x.To)
// .Any(apir => apir.Id == r.Id && (apir.From.Id != r.FromId || apir.To.Id != r.ToId))))
// {
// var existing = apiCtx.UserRoutes.Include(x => x.From).Include(x => x.To).Single(x => x.Id == route.Id);
// existing.From = apiCtx.Users.Single(x => x.Id == route.FromId);
// existing.To = apiCtx.Users.Single(x => x.Id == route.ToId);
// }
foreach (var deletedRoute in apiCtx.UserRoutes
.Where(x => !usersResponse.UserRoutes.Any(r => r.Id == x.Id)))
{
apiCtx.UserRoutes.Remove(deletedRoute);
}
// foreach (var deletedRoute in apiCtx.UserRoutes
// .Where(x => !usersResponse.UserRoutes.Any(r => r.Id == x.Id)))
// {
// apiCtx.UserRoutes.Remove(deletedRoute);
// }
foreach (var addedRoute in usersResponse.UserRoutes
.Where(r => !apiCtx.UserRoutes.Any(x => x.Id == r.Id)))
{
apiCtx.UserRoutes.Add(new MessengerApi.Db.Entities.UserRoute
{
Id = addedRoute.Id,
From = apiCtx.Users.Single(x => x.Id == addedRoute.FromId),
To = apiCtx.Users.Single(x => x.Id == addedRoute.ToId)
});
}
// foreach (var addedRoute in usersResponse.UserRoutes
// .Where(r => !apiCtx.UserRoutes.Any(x => x.Id == r.Id)))
// {
// apiCtx.UserRoutes.Add(new MessengerApi.Db.Entities.UserRoute
// {
// Id = addedRoute.Id,
// From = apiCtx.Users.Single(x => x.Id == addedRoute.FromId),
// To = apiCtx.Users.Single(x => x.Id == addedRoute.ToId)
// });
// }
broCtx.SaveChanges();
apiCtx.SaveChanges();
}
// broCtx.SaveChanges();
// apiCtx.SaveChanges();
// }
var messagesRequest = new HttpRequestMessage(HttpMethod.Get, $"{masterServer.BrokerApiUrl}/messages");
messagesRequest.Headers.Authorization = new System.Net.Http.Headers.AuthenticationHeaderValue("Bearer", this.settings.BrokerId.ToString());
// var messagesRequest = new HttpRequestMessage(HttpMethod.Get, $"{masterServer.BrokerApiUrl}/messages");
// messagesRequest.Headers.Authorization = new System.Net.Http.Headers.AuthenticationHeaderValue("Bearer", this.settings.BrokerId.ToString());
var messagesResponseMessage = await this.httpClient.SendAsync(messagesRequest, ct);
// var messagesResponseMessage = await this.httpClient.SendAsync(messagesRequest, ct);
if (!messagesResponseMessage.IsSuccessStatusCode)
{
continue;
}
// if (!messagesResponseMessage.IsSuccessStatusCode)
// {
// continue;
// }
var messagesResponse = JsonSerializer.Deserialize<Model.Http.Sync.SyncResponse>(await messagesResponseMessage.Content.ReadAsStringAsync());
// var messagesResponse = JsonSerializer.Deserialize<Model.Http.Messages.MessagesResponse>(await messagesResponseMessage.Content.ReadAsStringAsync());
using (var broCtx = new BrokerDbContext(this.settings.MessengerBrokerDbConnectionString))
using (var apiCtx = new MessengerDbContext(this.settings.MessengerApiDbConnectionString))
{
var addedMessages = messagesResponse.Messages
.Where(m => broCtx.Messages.Any(x => x.BrokerId == masterServer.BrokerId && x.Id == m.Id) == false)
.ToList();
// using (var broCtx = new BrokerDbContext(this.settings.MessengerBrokerDbConnectionString))
// using (var apiCtx = new MessengerDbContext(this.settings.MessengerApiDbConnectionString))
// {
// var addedMessages = messagesResponse.Messages
// .Where(m => broCtx.Messages.Any(x => x.BrokerId == masterServer.BrokerId && x.Id == m.Id) == false)
// .ToList();
addedMessages.ForEach(x =>
{
broCtx.Messages.Add(new Db.Model.Message
{
BrokerId = masterServer.BrokerId,
Id = x.Id
});
// addedMessages.ForEach(x =>
// {
// broCtx.Messages.Add(new Db.Model.Message
// {
// BrokerId = masterServer.BrokerId,
// Id = x.Id
// });
apiCtx.Messages.Add(new MessengerApi.Db.Entities.Message
{
Id = x.Id,
CreatedUtc = x.CreatedUtc,
From = x.From != null ? (apiCtx.Users.SingleOrDefault(u => u.Id == x.From.Value)) ?? null : null,
To = x.To != null ? (apiCtx.Users.SingleOrDefault(u => u.Id == x.To.Value)) ?? null : null,
IsAcknowledged = x.IsAcknowledged,
IsDelivered = x.IsDelivered,
Payload = x.Payload,
PayloadId = x.PayloadId,
PayloadLifespanInSeconds = x.PayloadLifespanInSeconds,
PayloadTimestamp = x.PayloadTimestamp,
PayloadType = x.PayloadType,
});
});
// apiCtx.Messages.Add(new MessengerApi.Db.Entities.Message
// {
// Id = x.Id,
// CreatedUtc = x.CreatedUtc,
// From = x.From != null ? (apiCtx.Users.SingleOrDefault(u => u.Id == x.From.Value)) ?? null : null,
// To = x.To != null ? (apiCtx.Users.SingleOrDefault(u => u.Id == x.To.Value)) ?? null : null,
// IsAcknowledged = x.IsAcknowledged,
// IsDelivered = x.IsDelivered,
// Payload = x.Payload,
// PayloadId = x.PayloadId,
// PayloadLifespanInSeconds = x.PayloadLifespanInSeconds,
// PayloadTimestamp = x.PayloadTimestamp,
// PayloadType = x.PayloadType,
// });
// });
var existingMessages = messagesResponse.Messages
.Except(addedMessages)
.ToList();
// var existingMessages = messagesResponse.Messages
// .Except(addedMessages)
// .ToList();
existingMessages.ForEach(x =>
{
var existing = apiCtx.Messages.SingleOrDefault(a => a.Id == x.Id);
// existingMessages.ForEach(x =>
// {
// var existing = apiCtx.Messages.SingleOrDefault(a => a.Id == x.Id);
if(existing != null && (existing.IsDelivered != x.IsDelivered || existing.IsAcknowledged != x.IsAcknowledged))
{
existing.IsDelivered = x.IsDelivered;
existing.IsAcknowledged = x.IsAcknowledged;
}
});
// if(existing != null && (existing.IsDelivered != x.IsDelivered || existing.IsAcknowledged != x.IsAcknowledged))
// {
// existing.IsDelivered = x.IsDelivered;
// existing.IsAcknowledged = x.IsAcknowledged;
// }
// });
broCtx.SaveChanges();
apiCtx.SaveChanges();
}
}
}
// broCtx.SaveChanges();
// apiCtx.SaveChanges();
// }
// }
// }
private async Task RemoveUser(Guid id, BrokerDbContext broCtx, MessengerDbContext apiCtx)
{
var broUser = await broCtx.Users.SingleOrDefaultAsync(x => x.Id == id);
// private async Task RemoveUser(Guid id, BrokerDbContext broCtx, MessengerDbContext apiCtx)
// {
// var broUser = await broCtx.Users.SingleOrDefaultAsync(x => x.Id == id);
if (broUser != null)
{
broCtx.Users.Remove(broUser);
}
// if (broUser != null)
// {
// broCtx.Users.Remove(broUser);
// }
var apiUser = await apiCtx.Users.SingleOrDefaultAsync(x => x.Id == id);
// var apiUser = await apiCtx.Users.SingleOrDefaultAsync(x => x.Id == id);
if (apiUser != null)
{
apiCtx.Users.Remove(apiUser);
}
}
// if (apiUser != null)
// {
// apiCtx.Users.Remove(apiUser);
// }
// }
private async Task AddUser(Model.Http.Users.UsersResponse.User user, Guid brokerId, BrokerDbContext broCtx, MessengerDbContext apiCtx)
{
if (broCtx.Users.GetUserExists(user.Id) == false)
{
broCtx.Users.Add(new Db.Model.User
{
Id = user.Id,
BrokerId = brokerId
});
}
// private async Task AddUser(Model.Http.Users.UsersResponse.User user, Guid brokerId, BrokerDbContext broCtx, MessengerDbContext apiCtx)
// {
// if (broCtx.Users.GetUserExists(user.Id) == false)
// {
// broCtx.Users.Add(new Db.Model.User
// {
// Id = user.Id,
// BrokerId = brokerId
// });
// }
if (apiCtx.Users.Any(x => x.Id == user.Id))
{
await this.UpdateUser(user, apiCtx);
}
else
{
await apiCtx.Users.AddAsync(new MessengerApi.Db.Entities.User
{
Id = user.Id,
ApiKey = user.ApiKey,
CanReceive = user.CanReceive,
CanSend = user.CanSend,
IsEnabled = user.IsEnabled,
Name = user.Name
});
}
}
// if (apiCtx.Users.Any(x => x.Id == user.Id))
// {
// await this.UpdateUser(user, apiCtx);
// }
// else
// {
// await apiCtx.Users.AddAsync(new MessengerApi.Db.Entities.User
// {
// Id = user.Id,
// ApiKey = user.ApiKey,
// CanReceive = user.CanReceive,
// CanSend = user.CanSend,
// IsEnabled = user.IsEnabled,
// Name = user.Name
// });
// }
// }
private async Task UpdateUser(Model.Http.Users.UsersResponse.User user, MessengerDbContext apiCtx)
{
var apiUser = await apiCtx.Users.SingleAsync(x => x.Id == user.Id);
apiUser.ApiKey = user.ApiKey;
apiUser.CanReceive = user.CanReceive;
apiUser.CanSend = user.CanSend;
apiUser.IsEnabled = user.IsEnabled;
apiUser.Name = user.Name;
}
}
}
// private async Task UpdateUser(Model.Http.Users.UsersResponse.User user, MessengerDbContext apiCtx)
// {
// var apiUser = await apiCtx.Users.SingleAsync(x => x.Id == user.Id);
// apiUser.ApiKey = user.ApiKey;
// apiUser.CanReceive = user.CanReceive;
// apiUser.CanSend = user.CanSend;
// apiUser.IsEnabled = user.IsEnabled;
// apiUser.Name = user.Name;
// }
// }
//}

4
code/MessengerBroker/MessengerBroker.csproj
View File

@ -12,7 +12,9 @@
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\sub\messengerapi\code\MessengerApi.Db\MessengerApi.Db.csproj" />
<ProjectReference Include="..\..\subm\messengerapi\code\MessengerApi\MessengerApi.csproj" />
<ProjectReference Include="..\MessengerBroker.Configuration\MessengerBroker.Configuration.csproj" />
<ProjectReference Include="..\MessengerBroker.Db.Sql\MessengerBroker.Db.Sql.csproj" />
<ProjectReference Include="..\MessengerBroker.Db\MessengerBroker.Db.csproj" />
</ItemGroup>

6
code/MessengerBroker/MessengerBroker.http
View File

@ -1,6 +0,0 @@
@MessengerBroker_HostAddress = http://localhost:5048
GET {{MessengerBroker_HostAddress}}/weatherforecast/
Accept: application/json
###

42
code/MessengerBroker/Model/Http/Sync.cs
View File

@ -1,42 +0,0 @@
namespace MessengerBroker.Model.Http
{
public class Sync
{
public class SyncRequest
{
public Guid BrokerId { get; set; }
public DateTime SinceUtc { get; set; }
}
public class SyncResponse
{
public Message[] Messages { get; set; }
public class Message
{
public Guid Id { get; set; }
public DateTime CreatedUtc { get; set; }
public Guid? From { get; set; }
public Guid? To { get; set; }
public bool IsDelivered { get; set; }
public bool IsAcknowledged { get; set; }
public string PayloadId { get; set; }
public string PayloadType { get; set; }
public string Payload { get; set; }
public DateTime? PayloadTimestamp { get; set; }
public int? PayloadLifespanInSeconds { get; set; }
}
}
}
}

12
code/MessengerBroker/Models/CachedIdentity.cs Normal file
View File

@ -0,0 +1,12 @@
using MessengerBroker.Configuration.Model.Servers;
using System.Security.Claims;
namespace MessengerBroker.Models
{
public class CachedIdentity
{
public SlaveServer Server { get; set; }
public ClaimsPrincipal ClaimsPrincipal { get; set; }
}
}

17
code/MessengerBroker/Models/Http/Messages.cs Normal file
View File

@ -0,0 +1,17 @@
namespace MessengerBroker.Model.Http
{
public class Messages
{
public class MessagesRequest
{
public Guid OwnerBrokerId { get; set; }
public DateTime SinceUtc { get; set; }
}
public class MessagesResponse
{
public MessengerApi.Db.Entities.Message[] Messages { get; set; }
}
}
}

4
code/MessengerBroker/Model/Http/Users.cs → code/MessengerBroker/Models/Http/Users.cs
View File

@ -17,10 +17,6 @@
public string Name { get; set; }
public bool IsEnabled { get; set; }
public bool CanSend { get; set; }
public bool CanReceive { get; set; }
}
public class UserRoute

9
code/MessengerBroker/Models/Scoped/Identity.cs Normal file
View File

@ -0,0 +1,9 @@
using MessengerBroker.Configuration.Model.Servers;
namespace MessengerBroker.Models.Scoped
{
public class Identity
{
public SlaveServer Server { get; set; }
}
}

179
code/MessengerBroker/Program.cs
View File

@ -1,6 +1,14 @@
using MessengerApi.Configuration.Sources.Environment;
using MessengerBroker.Configuration.Model;
using MessengerBroker.Factories;
using MessengerBroker.Handlers;
using MessengerBroker.Handlers.Endpoint;
using MessengerBroker.Model.Http;
using System.Runtime.CompilerServices;
using MessengerBroker.Models.Scoped;
using Microsoft.AspNetCore.Authentication;
using Microsoft.AspNetCore.HttpOverrides;
using Microsoft.EntityFrameworkCore;
using System.Net;
namespace MessengerBroker
{
@ -8,94 +16,119 @@ namespace MessengerBroker
{
public static void Main(string[] args)
{
var settings = (Settings)null;
BrokerConfiguration configuration = null;
try
{
configuration = new BrokerConfiguration(new EnvironmentConfigurationSource());
}
catch (Exception ex)
{
Console.WriteLine("Can't load settings.", ex);
throw;
}
var builder = WebApplication.CreateBuilder(args);
builder.Services.AddMemoryCache();
builder.Services.AddSingleton<ILogger, ConsoleLogger>();
builder.Services.AddSingleton<BrokerConfiguration>(configuration);
builder.Services.AddSingleton<BrokerDbContextFactory>();
builder.Services.AddSingleton<MessengerApi.Factories.DbContextFactory>((sp) =>
{
return new MessengerApi.Factories.DbContextFactory(configuration.ApiPersistenceConfiguration);
});
builder.Services.AddScoped<Identity>();
builder.Services.AddScoped<UsersEndpointHandler>();
builder.Services.AddScoped<MessagesEndpointHandler>();
// Authentication.
builder.Services
.AddAuthentication("Bearer")
.AddScheme<AuthenticationSchemeOptions, CustomBearerAuthenticationHandler>("Bearer", null);
// Proxy registration to forward real client IPs.
builder.Services.Configure<ForwardedHeadersOptions>(options =>
{
options.ForwardedHeaders = ForwardedHeaders.XForwardedFor | ForwardedHeaders.XForwardedProto;
foreach (var proxy in configuration.Proxies)
{
options.KnownProxies.Add(IPAddress.Parse(proxy));
}
});
var app = builder.Build();
app.MapGet("/users", async (HttpContext httpContext) =>
// DB Migrations
using (var ctx = app.Services.GetRequiredService<BrokerDbContextFactory>().CreateDbContext())
{
var authHandler = app.Services.GetService<AuthHandler>();
var brokerId = authHandler.Auth(httpContext);
var migrationLogger = app.Services.GetRequiredService<ILogger>();
if(brokerId == null)
try
{
return Results.Unauthorized();
if (ctx.Database.GetPendingMigrations().Any())
{
migrationLogger.Info("Applying migrations.");
ctx.Database.Migrate();
}
else
{
migrationLogger.Info("No migrations pending.");
}
}
var dataHandler = app.Services.GetService<DataHandler>();
var usersAndRoutes = await dataHandler.GetLocalUsersAndRoutes();
var response = new Users.UsersResponse
catch (Exception ex)
{
Users = usersAndRoutes.Item1.Select(x => new Users.UsersResponse.User
{
Id = x.Id,
Name = x.Name,
ApiKey = x.ApiKey,
CanReceive = x.CanReceive,
CanSend = x.CanSend,
IsEnabled = x.IsEnabled,
}).ToArray(),
UserRoutes = usersAndRoutes.Item2.Select(x => new Users.UsersResponse.UserRoute
{
Id = x.Id,
FromId = x.From.Id,
ToId = x.To.Id
}).ToArray()
};
migrationLogger.Error("Can't run migrations successfully.", ex);
throw;
}
}
//// Housekeeping.
//if (configuration.HousekeepingEnabled)
//{
// _ = Task.Run(async () =>
// {
// while (true)
// {
// await app.Services.GetService<HousekeepingHandler>().RemoveOldMessages();
// await Task.Delay(TimeSpan.FromMinutes(1));
// }
// });
//}
//// Run pull sync from masters.
//_ = Task.Run(async () =>
//{
// var cts = new CancellationTokenSource();
// var handler = app.Services.GetService<MasterHandler>();
// foreach (var master in settings.Masters)
// {
// _ = handler.BeginSyncingWithMaster(master, cts.Token);
// }
//});
app.UseStaticFiles();
app.UseRouting();
app.UseAuthentication();
app.UseForwardedHeaders();
app.MapGet("/users", async (UsersEndpointHandler handler) =>
{
var response = await handler.GetUsers();
return Results.Json(response);
});
app.MapGet("/sync", async (HttpContext httpContext, [AsParameters] Sync.SyncRequest request) =>
app.MapGet("/messages", async (
MessagesEndpointHandler handler,
[AsParameters] Messages.MessagesRequest request) =>
{
var authHandler = app.Services.GetService<AuthHandler>();
var brokerId = authHandler.Auth(httpContext);
if(brokerId == null)
{
return Results.Unauthorized();
}
else if(request.BrokerId != brokerId && request.BrokerId != settings.BrokerId)
{
return Results.Unauthorized();
}
var dataHandler = app.Services.GetService<DataHandler>();
var messages = await dataHandler.GetMessages(request.BrokerId, request.SinceUtc);
var response = new Sync.SyncResponse
{
Messages = messages.Select(x => new Sync.SyncResponse.Message
{
Id = x.Id,
CreatedUtc = x.CreatedUtc,
From = x.From.Id,
To = x.To.Id,
IsAcknowledged = x.IsAcknowledged,
IsDelivered = x.IsDelivered,
Payload = x.Payload,
PayloadId = x.PayloadId,
PayloadLifespanInSeconds = x.PayloadLifespanInSeconds,
PayloadTimestamp = x.PayloadTimestamp,
PayloadType = x.PayloadType
}).ToArray()
};
var response = await handler.GetMessages(request);
return Results.Json(response);
});
_ = Task.Run(async () =>
{
var cts = new CancellationTokenSource();
var handler = app.Services.GetService<MasterHandler>();
foreach(var master in settings.Masters)
{
_ = handler.BeginSyncingWithMaster(master, cts.Token);
}
});
app.Run();
}
}

36
code/MessengerBroker/Settings.cs
View File

@ -1,36 +0,0 @@
namespace MessengerBroker
{
public class Settings
{
/// <summary>
/// Connection string to Messenger API DB.
/// </summary>
public string MessengerApiDbConnectionString { get; set; }
public string MessengerBrokerDbConnectionString { get; set; }
public Guid BrokerId { get; set; }
public MasterServer[] Masters { get; set; }
public SlaveServer[] Slaves { get; set; }
/// <summary>
/// A server that we are a slave to. If this server goes down, their users will alternate to us and we have to provide service during outage. We pull data from this server.
/// </summary>
public class MasterServer
{
public string BrokerApiUrl { get; set; }
public Guid BrokerId { set; get; }
}
/// <summary>
/// A server that slaves to us in case of our own outage. They pull from us.
/// </summary>
public class SlaveServer
{
public Guid BrokerId { get; set; }
}
}
}

8
code/MessengerBroker/appsettings.Development.json
View File

@ -1,8 +0,0 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.AspNetCore": "Warning"
}
}
}

9
code/MessengerBroker/appsettings.json
View File

@ -1,9 +0,0 @@
{
"Logging": {
"LogLevel": {
"Default": "Information",
"Microsoft.AspNetCore": "Warning"
}
},
"AllowedHosts": "*"
}