Initial commit HEAD master

8 files changed, 634 insertions, 0 deletions

diff --git a/MatrixContentFilter/Services/AsyncActionQueues/AbstractionAsyncActionQueue.cs b/MatrixContentFilter/Services/AsyncActionQueues/AbstractionAsyncActionQueue.cs
new file mode 100644
index 0000000..f4c559c
--- /dev/null
+++ b/MatrixContentFilter/Services/AsyncActionQueues/AbstractionAsyncActionQueue.cs
@@ -0,0 +1,38 @@
+using System.Collections.Concurrent;
+using System.Threading.Channels;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+
+namespace MatrixContentFilter.Services.AsyncActionQueues;
+
+public abstract class AbstractAsyncActionQueue : BackgroundService {
+    private readonly ConcurrentStack<string> _recentIds = new();
+    private readonly Channel<Func<Task>> _queue = Channel.CreateUnbounded<Func<Task>>(new UnboundedChannelOptions() {
+        SingleReader = true
+    });
+    private static CancellationTokenSource _cts = new();
+
+    /// <summary>
+    ///     Enqueue an action to be executed asynchronously
+    /// </summary>
+    /// <param name="id">Reproducible ID</param>
+    /// <param name="action">Action to execute</param>
+    /// <returns>`true` if action was appended, `false` if action was not added, eg. due to duplicate ID</returns>
+    public virtual async Task<bool> EqueueActionAsync(string id, Func<Task> action) {
+        throw new NotImplementedException();
+    }
+
+    private async Task ProcessQueue() {
+        throw new NotImplementedException();
+    }
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken) {
+        while (!stoppingToken.IsCancellationRequested) {
+            await ProcessQueue();
+            Console.WriteLine("AbstractAsyncActionQueue waiting for new actions, this should never happen!");
+        }
+
+        //clear backlog and exit
+        await ProcessQueue();
+    }
+}
\ No newline at end of file
diff --git a/MatrixContentFilter/Services/AsyncActionQueues/FiFoAsyncActionQueue.cs b/MatrixContentFilter/Services/AsyncActionQueues/FiFoAsyncActionQueue.cs
new file mode 100644
index 0000000..3d7c90d
--- /dev/null
+++ b/MatrixContentFilter/Services/AsyncActionQueues/FiFoAsyncActionQueue.cs
@@ -0,0 +1,60 @@
+using System.Collections.Concurrent;
+using System.Threading.Channels;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+
+namespace MatrixContentFilter.Services.AsyncActionQueues;
+
+public class FiFoAsyncActionQueue(ILogger<FiFoAsyncActionQueue> logger, MatrixContentFilterConfiguration cfg) : AbstractAsyncActionQueue {
+    // private readonly ConcurrentQueue<(string Id, Func<Task> Action)> _queue = new();
+    private readonly HashSet<string> _recentIds = new();
+    private readonly Channel<Func<Task>> _queue = Channel.CreateUnbounded<Func<Task>>(new UnboundedChannelOptions() {
+        SingleReader = true
+    });
+    private readonly SemaphoreSlim _semaphore = new(cfg.ConcurrencyLimits.Redactions, cfg.ConcurrencyLimits.Redactions);
+
+    /// <summary>
+    ///     Enqueue an action to be executed asynchronously
+    /// </summary>
+    /// <param name="id">Reproducible ID</param>
+    /// <param name="action">Action to execute</param>
+    /// <returns>`true` if action was appended, `false` if action was not added, eg. due to duplicate ID</returns>
+    public override async Task<bool> EqueueActionAsync(string id, Func<Task> action) {
+        if (_recentIds.Contains(id)) {
+            logger.LogWarning("Duplicate action ID detected, ignoring action");
+            return false;
+        }
+        await _queue.Writer.WriteAsync(action);
+        _recentIds.Add(id);
+
+        if (_queue.Reader.Count > 100) {
+            logger.LogWarning("Action Queue is getting full, consider increasing the rate limit or exempting the bot!");
+        }
+
+        return true;
+    }
+
+    private async Task ProcessQueue() {
+        await foreach (var task in _queue.Reader.ReadAllAsync()) {
+            await _semaphore.WaitAsync();
+            _ = Task.Run(async () => {
+                try {
+                    await task.Invoke();
+                }
+                finally {
+                    _semaphore.Release();
+                }
+            });
+        }
+    }
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken) {
+        while (!stoppingToken.IsCancellationRequested) {
+            await ProcessQueue();
+            logger.LogWarning("Waiting for new actions, ProcessQueue returned early!");
+        }
+
+        //clear backlog and exit
+        await ProcessQueue();
+    }
+}
\ No newline at end of file
diff --git a/MatrixContentFilter/Services/AsyncActionQueues/LiFoAsyncActionQueue.cs b/MatrixContentFilter/Services/AsyncActionQueues/LiFoAsyncActionQueue.cs
new file mode 100644
index 0000000..631cc74
--- /dev/null
+++ b/MatrixContentFilter/Services/AsyncActionQueues/LiFoAsyncActionQueue.cs
@@ -0,0 +1,67 @@
+using System.Collections.Concurrent;
+using Microsoft.Extensions.Logging;
+
+namespace MatrixContentFilter.Services.AsyncActionQueues;
+
+public class LiFoAsyncActionQueue(ILogger<LiFoAsyncActionQueue> logger, MatrixContentFilterConfiguration cfg) : AbstractAsyncActionQueue {
+    // private readonly ConcurrentQueue<(string Id, Func<Task> Action)> _queue = new();
+    private readonly HashSet<string> _recentIds = new();
+    private readonly ConcurrentStack<(string Id, Func<Task> Action)> _queue = new();
+    private static CancellationTokenSource _cts = new();
+    private readonly SemaphoreSlim _semaphore = new(cfg.ConcurrencyLimits.Redactions, cfg.ConcurrencyLimits.Redactions);
+
+    /// <summary>
+    ///     Enqueue an action to be executed asynchronously
+    /// </summary>
+    /// <param name="id">Reproducible ID</param>
+    /// <param name="action">Action to execute</param>
+    /// <returns>`true` if action was appended, `false` if action was not added, eg. due to duplicate ID</returns>
+    public override async Task<bool> EqueueActionAsync(string id, Func<Task> action) {
+        if (_recentIds.Contains(id)) {
+            logger.LogWarning("Duplicate action ID detected, ignoring action");
+            return false;
+        }
+
+        _queue.Push((id, action));
+        _recentIds.Add(id);
+        _cts.Cancel(false);
+
+        return true;
+    }
+
+    private async Task ProcessQueue() {
+        // await foreach (var task in _queue2.Reader.ReadAllAsync()) {
+        while (_queue.TryPop(out var task)) {
+            await _semaphore.WaitAsync();
+            _ = Task.Run(async () => {
+                try {
+                    await task.Action.Invoke();
+                    _recentIds.Remove(task.Id);
+                }
+                finally {
+                    _semaphore.Release();
+                }
+            });
+        }
+
+        _cts.Dispose();
+        _cts = new CancellationTokenSource();
+    }
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken) {
+        while (!stoppingToken.IsCancellationRequested) {
+            await ProcessQueue();
+            Console.WriteLine(GetType().Name + " waiting for new actions");
+            try {
+                await Task.Delay(10000, _cts.Token);
+            }
+            catch (TaskCanceledException) {
+                Console.WriteLine(GetType().Name + " _cts cancelled");
+                // ignore
+            }
+        }
+
+        //clear backlog and exit
+        await ProcessQueue();
+    }
+}
\ No newline at end of file
diff --git a/MatrixContentFilter/Services/AsyncMessageQueue.cs b/MatrixContentFilter/Services/AsyncMessageQueue.cs
new file mode 100644
index 0000000..7912cc5
--- /dev/null
+++ b/MatrixContentFilter/Services/AsyncMessageQueue.cs
@@ -0,0 +1,43 @@
+using System.Collections.Concurrent;
+using LibMatrix.EventTypes.Spec;
+using LibMatrix.RoomTypes;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+
+namespace MatrixContentFilter.Services;
+
+public class AsyncMessageQueue(ILogger<AsyncMessageQueue> logger, MatrixContentFilterConfiguration cfg) : BackgroundService {
+    private readonly ConcurrentQueue<(GenericRoom Room, RoomMessageEventContent Content)> _queue = new();
+    private readonly SemaphoreSlim _semaphore = new(cfg.ConcurrencyLimits.LogMessages, cfg.ConcurrencyLimits.LogMessages);
+    public void EnqueueMessageAsync(GenericRoom room, RoomMessageEventContent content) {
+        _queue.Enqueue((room, content));
+        
+        if (_queue.Count > 100) {
+            logger.LogWarning($"Message Queue is getting full (c={_queue.Count}), consider increasing the rate limit or exempting the bot!");
+        }
+    }
+    
+    private async Task ProcessQueue() {
+        while (_queue.TryDequeue(out var message)) {
+            await _semaphore.WaitAsync();
+            _ = Task.Run(async () => {
+                try {
+                    await message.Room.SendMessageEventAsync(message.Content);
+                }
+                finally {
+                    _semaphore.Release();
+                }
+            });
+        }
+    }
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken) {
+        while (!stoppingToken.IsCancellationRequested) {
+            await ProcessQueue();
+            await Task.Delay(1000, stoppingToken);
+        }
+        
+        //clear backlog and exit
+        await ProcessQueue();
+    }
+}
\ No newline at end of file
diff --git a/MatrixContentFilter/Services/BotModeSanityCheckService.cs b/MatrixContentFilter/Services/BotModeSanityCheckService.cs
new file mode 100644
index 0000000..55fe9e8
--- /dev/null
+++ b/MatrixContentFilter/Services/BotModeSanityCheckService.cs
@@ -0,0 +1,56 @@
+using System.Diagnostics;
+using ArcaneLibs;
+using ArcaneLibs.Extensions;
+using LibMatrix;
+using LibMatrix.Filters;
+using LibMatrix.Helpers;
+using LibMatrix.Homeservers;
+using LibMatrix.Responses;
+using MatrixContentFilter.Abstractions;
+using MatrixContentFilter.Handlers.Filters;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+
+namespace MatrixContentFilter.Services;
+
+public class BotModeSanityCheckService(
+    ILogger<BotModeSanityCheckService> logger,
+    AuthenticatedHomeserverGeneric hs,
+    ConfigurationService filterConfigService,
+    IEnumerable<IContentFilter> filters,
+    AsyncMessageQueue msgQueue
+) : BackgroundService {
+    /// <summary>Triggered when the application host is ready to start the service.</summary>
+    /// <param name="cancellationToken">Indicates that the start process has been aborted.</param>
+    protected override async Task ExecuteAsync(CancellationToken cancellationToken) {
+        while (!cancellationToken.IsCancellationRequested) {
+            await Task.Delay(10000, cancellationToken);
+            var rooms = await hs.GetJoinedRooms();
+            rooms.RemoveAll(x => x.RoomId == filterConfigService.LogRoom.RoomId);
+            rooms.RemoveAll(x => x.RoomId == filterConfigService.ControlRoom.RoomId);
+
+            var timelineFilter = new SyncFilter.RoomFilter.StateFilter(notTypes: ["m.room.redaction"], limit: 5000);
+            var timelines = rooms.Select(async x => {
+                var room = hs.GetRoom(x.RoomId);
+                // var sync = await room.GetMessagesAsync(null, 1500, filter: timelineFilter.ToJson(ignoreNull: true, indent: false).UrlEncode());
+                var iter = room.GetManyMessagesAsync(null, 5000, filter: timelineFilter.ToJson(ignoreNull: true, indent: false).UrlEncode(), chunkSize: 250);
+                await foreach (var sync in iter) {
+                    var tasks = Parallel.ForEachAsync(filters, async (filter, ct) => {
+                        try {
+                            Console.WriteLine("Processing filter {0} (sanity check, chunk[s={1}])", filter.GetType().FullName, sync.Chunk.Count);
+                            await filter.ProcessEventListAsync(sync.Chunk);
+                        }
+                        catch (Exception e) {
+                            logger.LogError(e, "Error processing sync with filter {filter}", filter.GetType().FullName);
+                            msgQueue.EnqueueMessageAsync(filterConfigService.LogRoom, new MessageBuilder("m.notice")
+                                .WithBody($"Error processing sync with filter {filter.GetType().FullName}: {e.Message}").Build());
+                        }
+                    });
+
+                    await tasks;
+                }
+            }).ToList();
+            await Task.WhenAll(timelines);
+        }
+    }
+}
\ No newline at end of file
diff --git a/MatrixContentFilter/Services/ConfigurationService.cs b/MatrixContentFilter/Services/ConfigurationService.cs
new file mode 100644
index 0000000..f83c89a
--- /dev/null
+++ b/MatrixContentFilter/Services/ConfigurationService.cs
@@ -0,0 +1,198 @@
+using ArcaneLibs.Extensions;
+using LibMatrix;
+using LibMatrix.EventTypes.Spec.State;
+using LibMatrix.Helpers;
+using LibMatrix.Homeservers;
+using LibMatrix.Responses;
+using LibMatrix.RoomTypes;
+using LibMatrix.Utilities;
+using MatrixContentFilter.EventTypes;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+
+namespace MatrixContentFilter.Services;
+
+public class ConfigurationService(ILogger<ConfigurationService> logger, AuthenticatedHomeserverGeneric hs, AsyncMessageQueue msgQueue) : BackgroundService {
+    public BotEnvironmentConfiguration EnvironmentConfiguration { get; private set; }
+    public FilterConfiguration DefaultConfiguration { get; private set; }
+    public Dictionary<string, FilterConfiguration> RoomConfigurationOverrides { get; } = new();
+    public Dictionary<string, FilterConfiguration> FinalRoomConfigurations { get; } = new();
+
+    public GenericRoom LogRoom { get; private set; } = null!;
+    public GenericRoom ControlRoom { get; private set; } = null!;
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken) {
+        var syncHelper = new SyncHelper(hs, logger) {
+            NamedFilterName = CommonSyncFilters.GetAccountDataWithRooms
+        };
+
+        await foreach (var sync in syncHelper.EnumerateSyncAsync(stoppingToken).WithCancellation(stoppingToken)) {
+            if (sync is { AccountData: null, Rooms: null }) continue;
+            logger.LogInformation("Received configuration update: {syncData}", sync.ToJson(ignoreNull: true));
+            await OnSyncReceived(sync);
+        }
+    }
+
+    public async Task OnSyncReceived(SyncResponse sync) {
+        if (sync.AccountData?.Events?.FirstOrDefault(x => x.Type == BotEnvironmentConfiguration.EventId) is { } envEvent) {
+            EnvironmentConfiguration = envEvent.TypedContent as BotEnvironmentConfiguration;
+            msgQueue.EnqueueMessageAsync(LogRoom, new MessageBuilder("m.notice")
+                .WithColoredBody("#FF0088", "Environment configuration updated from sync.").WithNewline()
+                .WithCollapsibleSection("JSON data:", msb => msb.WithCodeBlock(EnvironmentConfiguration.ToJson(), "json"))
+                // .WithCollapsibleSection("Full event JSON", _msb => _msb.WithCodeBlock(envEvent.ToJson(), "json"))
+                .Build());
+            LogRoom = hs.GetRoom(EnvironmentConfiguration.LogRoomId!);
+            ControlRoom = hs.GetRoom(EnvironmentConfiguration.ControlRoomId!);
+        }
+
+        if (sync.AccountData?.Events?.FirstOrDefault(x => x.Type == FilterConfiguration.EventId) is { } filterEvent) {
+            DefaultConfiguration = filterEvent.TypedContent as FilterConfiguration;
+            msgQueue.EnqueueMessageAsync(LogRoom, new MessageBuilder("m.notice")
+                .WithColoredBody("#00FF88", "Default filter configuration updated from sync.").WithNewline()
+                .WithCollapsibleSection("JSON data:", msb => msb.WithCodeBlock(DefaultConfiguration.ToJson(), "json"))
+                // .WithCollapsibleSection("Full event JSON", _msb => _msb.WithCodeBlock(filterEvent.ToJson(), "json"))
+                .Build());
+        }
+
+        await Parallel.ForEachAsync(sync.Rooms?.Join ?? [], async (syncRoom, ct) => {
+            var (roomId, roomData) = syncRoom;
+            if (roomId == LogRoom!.RoomId || roomId == ControlRoom!.RoomId) return;
+            var room = hs.GetRoom(roomId);
+
+            if (roomData.AccountData?.Events?.FirstOrDefault(x => x.Type == FilterConfiguration.EventId) is { } roomFilterEvent) {
+                RoomConfigurationOverrides[roomId] = roomFilterEvent.TypedContent as FilterConfiguration;
+                var roomName = await room.GetNameOrFallbackAsync();
+                msgQueue.EnqueueMessageAsync(LogRoom, new MessageBuilder("m.notice")
+                    .WithColoredBody("#00FF88", msb => msb.WithBody($"Filter configuration updated for ").WithMention(roomId, roomName).WithBody(" from sync.")).WithNewline()
+                    .WithCollapsibleSection("JSON data:", msb => msb.WithCodeBlock(RoomConfigurationOverrides[roomId].ToJson(), "json"))
+                    .Build());
+            }
+        });
+    }
+
+    public async Task OnStartup(MatrixContentFilterConfiguration configuration) {
+        BotEnvironmentConfiguration _environmentConfiguration;
+        try {
+            _environmentConfiguration = await hs.GetAccountDataAsync<BotEnvironmentConfiguration>(BotEnvironmentConfiguration.EventId);
+        }
+        catch (MatrixException e) {
+            if (e is not { ErrorCode: MatrixException.ErrorCodes.M_NOT_FOUND }) throw;
+            logger.LogWarning("No environment configuration found, creating one");
+            _environmentConfiguration = new BotEnvironmentConfiguration();
+        }
+
+        if (string.IsNullOrWhiteSpace(_environmentConfiguration.ControlRoomId)) {
+            LogRoom = await hs.CreateRoom(new() {
+                Name = "MatrixContentFilter logs",
+                Invite = configuration.Admins,
+                Visibility = "private"
+            });
+            var powerlevels = await LogRoom.GetPowerLevelsAsync();
+            powerlevels.EventsDefault = 20;
+            foreach (var admin in configuration.Admins) {
+                powerlevels.Users[admin] = 100;
+            }
+
+            await LogRoom.SendStateEventAsync(RoomPowerLevelEventContent.EventId, powerlevels);
+
+            _environmentConfiguration.LogRoomId = LogRoom.RoomId;
+            await hs.SetAccountDataAsync(BotEnvironmentConfiguration.EventId, _environmentConfiguration);
+        }
+        else {
+            LogRoom = hs.GetRoom(_environmentConfiguration.LogRoomId!);
+        }
+
+        if (string.IsNullOrWhiteSpace(_environmentConfiguration.ControlRoomId)) {
+            ControlRoom = await hs.CreateRoom(new() {
+                Name = "MatrixContentFilter control room",
+                Invite = configuration.Admins,
+                Visibility = "private"
+            });
+            var powerlevels = await ControlRoom.GetPowerLevelsAsync();
+            powerlevels.EventsDefault = 20;
+            foreach (var admin in configuration.Admins) {
+                powerlevels.Users[admin] = 100;
+            }
+
+            await ControlRoom.SendStateEventAsync(RoomPowerLevelEventContent.EventId, powerlevels);
+
+            _environmentConfiguration.ControlRoomId = ControlRoom.RoomId;
+            await hs.SetAccountDataAsync(BotEnvironmentConfiguration.EventId, _environmentConfiguration);
+        }
+        else {
+            ControlRoom = hs.GetRoom(_environmentConfiguration.ControlRoomId!);
+        }
+
+        FilterConfiguration _filterConfiguration;
+        try {
+            _filterConfiguration = await hs.GetAccountDataAsync<FilterConfiguration>(FilterConfiguration.EventId);
+        }
+        catch (MatrixException e) {
+            if (e is not { ErrorCode: MatrixException.ErrorCodes.M_NOT_FOUND }) throw;
+            logger.LogWarning("No filter configuration found, creating one");
+            msgQueue.EnqueueMessageAsync(LogRoom, new MessageBuilder("m.notice").WithColoredBody("#FF0000", "No filter configuration found, creating one").Build());
+            _filterConfiguration = new FilterConfiguration();
+        }
+
+        Dictionary<string, object> changes = new();
+
+        T Log<T>(string key, T value) {
+            changes[key] = value;
+            return value;
+        }
+
+        _filterConfiguration.IgnoredUsers ??= Log("ignored_users", (List<string>) [
+            hs.WhoAmI.UserId,
+            .. configuration.Admins
+        ]);
+
+        _filterConfiguration.FileFilter ??= new();
+        _filterConfiguration.FileFilter.IgnoredUsers ??= Log("file_filter->ignored_users", (List<string>) []);
+        _filterConfiguration.FileFilter.Allowed ??= Log("file_filter->allowed", false);
+
+        _filterConfiguration.ImageFilter ??= new();
+        _filterConfiguration.ImageFilter.IgnoredUsers ??= Log("image_filter->ignored_users", (List<string>) []);
+        _filterConfiguration.ImageFilter.Allowed ??= Log("image_filter->allowed", false);
+
+        _filterConfiguration.VideoFilter ??= new();
+        _filterConfiguration.VideoFilter.IgnoredUsers ??= Log("video_filter->ignored_users", (List<string>) []);
+        _filterConfiguration.VideoFilter.Allowed ??= Log("video_filter->allowed", false);
+
+        _filterConfiguration.AudioFilter ??= new();
+        _filterConfiguration.AudioFilter.IgnoredUsers ??= Log("audio_filter->ignored_users", (List<string>) []);
+        _filterConfiguration.AudioFilter.Allowed ??= Log("audio_filter->allowed", false);
+
+        _filterConfiguration.UrlFilter ??= new();
+        _filterConfiguration.UrlFilter.IgnoredUsers ??= Log("url_filter->ignored_users", (List<string>) []);
+        _filterConfiguration.UrlFilter.Allowed ??= Log("url_filter->allowed", false);
+
+        if (changes.Count > 0) {
+            await hs.SetAccountDataAsync(FilterConfiguration.EventId, _filterConfiguration);
+            msgQueue.EnqueueMessageAsync(LogRoom, new MessageBuilder("m.notice").WithColoredBody("#FF0000", "Default filter configuration updated").WithNewline()
+                .WithTable(msb => {
+                    msb = msb.WithTitle("Default configuration changes", 2);
+
+                    foreach (var (key, value) in changes) {
+                        var formattedValue = value switch {
+                            List<string> list => string.Join(", ", list),
+                            _ => value.ToString()
+                        };
+                        msb = msb.WithRow(rb => { rb.WithCell(key).WithCell(formattedValue ?? "formattedValue was null!"); });
+                    }
+                }).Build());
+        }
+    }
+
+    private async Task RebuildRoomConfigurations(FilterConfiguration? defaultConfig, Dictionary<string, FilterConfiguration?>? roomConfigurations) {
+        defaultConfig ??= await hs.GetAccountDataAsync<FilterConfiguration>(FilterConfiguration.EventId);
+    }
+
+    public async Task<FilterConfiguration> GetFinalRoomConfiguration(string roomId) {
+        if (FinalRoomConfigurations.TryGetValue(roomId, out var config)) return config;
+        var roomConfig = RoomConfigurationOverrides.GetValueOrDefault(roomId);
+        var defaultConfig = DefaultConfiguration;
+
+        FinalRoomConfigurations[roomId] = config;
+        return config;
+    }
+}
\ No newline at end of file
diff --git a/MatrixContentFilter/Services/InfoCacheService.cs b/MatrixContentFilter/Services/InfoCacheService.cs
new file mode 100644
index 0000000..974e873
--- /dev/null
+++ b/MatrixContentFilter/Services/InfoCacheService.cs
@@ -0,0 +1,29 @@
+using ArcaneLibs.Collections;
+using LibMatrix.EventTypes.Spec.State;
+using LibMatrix.Homeservers;
+
+namespace MatrixContentFilter.Services;
+
+public class InfoCacheService(AuthenticatedHomeserverGeneric hs) {
+    private static readonly ExpiringSemaphoreCache<string> DisplayNameCache = new();
+    public static readonly ExpiringSemaphoreCache<string> RoomNameCache = new();
+
+    public async Task<string> GetDisplayNameAsync(string roomId, string userId) =>
+        await DisplayNameCache.GetOrAdd($"{roomId}\t{userId}", async () => {
+            var room = hs.GetRoom(roomId);
+            var userState = await room.GetStateAsync<RoomMemberEventContent>(RoomMemberEventContent.EventId, userId);
+            if (!string.IsNullOrWhiteSpace(userState?.DisplayName)) return userState.DisplayName;
+
+            var user = await hs.GetProfileAsync(userId);
+            if (!string.IsNullOrWhiteSpace(user?.DisplayName)) return user.DisplayName;
+
+            return userId;
+        }, TimeSpan.FromMinutes(5));
+
+    public async Task<string> GetRoomNameAsync(string roomId) =>
+        await RoomNameCache.GetOrAdd(roomId, async () => {
+            var room = hs.GetRoom(roomId);
+            var name = await room.GetNameOrFallbackAsync();
+            return name;
+        }, TimeSpan.FromMinutes(30));
+}
\ No newline at end of file
diff --git a/MatrixContentFilter/Services/MatrixContentFilterBot.cs b/MatrixContentFilter/Services/MatrixContentFilterBot.cs
new file mode 100644
index 0000000..321cdd4
--- /dev/null
+++ b/MatrixContentFilter/Services/MatrixContentFilterBot.cs
@@ -0,0 +1,143 @@
+using System.Diagnostics;
+using ArcaneLibs;
+using ArcaneLibs.Extensions;
+using LibMatrix.Filters;
+using LibMatrix.Helpers;
+using LibMatrix.Homeservers;
+using LibMatrix.Responses;
+using MatrixContentFilter.Abstractions;
+using MatrixContentFilter.Handlers.Filters;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+
+namespace MatrixContentFilter.Services;
+
+public class MatrixContentFilterBot(
+    ILogger<MatrixContentFilterBot> logger,
+    AuthenticatedHomeserverGeneric hs,
+    MatrixContentFilterConfiguration configuration,
+    ConfigurationService filterConfigService,
+    IEnumerable<IContentFilter> filters,
+    AsyncMessageQueue msgQueue
+) : BackgroundService {
+    /// <summary>Triggered when the application host is ready to start the service.</summary>
+    /// <param name="cancellationToken">Indicates that the start process has been aborted.</param>
+    protected override async Task ExecuteAsync(CancellationToken cancellationToken) {
+        try {
+            await filterConfigService.OnStartup(configuration);
+            msgQueue.EnqueueMessageAsync(filterConfigService.LogRoom, new MessageBuilder("m.notice").WithColoredBody("#00FF00", "Bot startup successful! Listening for events.")
+                .Build());
+            msgQueue.EnqueueMessageAsync(filterConfigService.LogRoom, new MessageBuilder("m.notice").WithColoredBody("#00FF00", msb => {
+                msb = msb.WithBody("Inserted filters implementations (internal):").WithNewline();
+                foreach (var filter in filters) {
+                    msb = msb.WithBody(filter.GetType().FullName).WithNewline();
+                }
+            }).Build());
+        }
+        catch (Exception e) {
+            logger.LogError(e, "Error on startup");
+            Environment.Exit(1); // We don't want to do a graceful shutdown if we can't start up
+        }
+
+        logger.LogInformation("Bot started!");
+        await Run(cancellationToken);
+    }
+
+    private SyncHelper syncHelper;
+
+    private async Task Run(CancellationToken cancellationToken) {
+        var syncFilter = new SyncFilter() {
+            Room = new() {
+                NotRooms = [filterConfigService.LogRoom.RoomId],
+                Timeline = new(notTypes: ["m.room.redaction"])
+            }
+        };
+        syncHelper = new SyncHelper(hs, logger) {
+            Filter = syncFilter
+        };
+        int i = 0;
+        await foreach (var sync in syncHelper.EnumerateSyncAsync(cancellationToken).WithCancellation(cancellationToken)) {
+            // if (i++ >= 100) {
+            //     var sw = Stopwatch.StartNew();
+            //     for (int gen = 0; gen < GC.MaxGeneration; gen++) {
+            //         GC.Collect(gen, GCCollectionMode.Forced, true, true);
+            //     }
+            //     i = 0;
+            //     msgQueue.EnqueueMessageAsync(filterConfigService.LogRoom, new MessageBuilder("m.notice")
+            //         .WithBody($"Garbage collection took {sw.ElapsedMilliseconds}ms")
+            //         .Build());
+            //     GC.
+            // }
+
+            // GC.TryStartNoGCRegion(1024 * 1024 * 1024);
+            var sw = Stopwatch.StartNew();
+            int actionCount = filters.Sum(x => x.ActionCount);
+            try {
+                await OnSyncReceived(sync);
+            }
+            catch (Exception e) {
+                logger.LogError(e, "Error processing sync");
+                msgQueue.EnqueueMessageAsync(filterConfigService.LogRoom, new MessageBuilder("m.notice").WithBody($"Error processing sync: {e.Message}").Build());
+            }
+            finally {
+                Console.WriteLine("Processed sync in {0}, executed {1} actions, {2} of memory usage", sw.Elapsed, filters.Sum(x => x.ActionCount) - actionCount, Util.BytesToString(Environment.WorkingSet));
+                // GC.EndNoGCRegion();
+            }
+            
+            // update sync filter
+            if (syncFilter.Room.NotRooms[0] != filterConfigService.LogRoom.RoomId) {
+                syncFilter.Room.NotRooms = [filterConfigService.LogRoom.RoomId];
+                syncHelper.Filter = syncFilter;
+            }
+        }
+    }
+
+    private int _syncCount;
+
+    private async Task OnSyncReceived(SyncResponse sync) {
+        if (_syncCount++ == 0) return; // Skip initial sync :/
+
+        if (sync.Rooms?.Join?.ContainsKey(filterConfigService.LogRoom.RoomId) == true) {
+            sync.Rooms?.Join?.Remove(filterConfigService.LogRoom.RoomId);
+        }
+
+        if (sync.Rooms?.Join?.ContainsKey(filterConfigService.ControlRoom.RoomId) == true) {
+            sync.Rooms?.Join?.Remove(filterConfigService.ControlRoom.RoomId);
+        }
+
+        // HACK: Server likes to send partial timelines during elevated activity, so we need to fetch them in order not to miss events
+        var timelineFilter = new SyncFilter.RoomFilter.StateFilter(notTypes: ["m.room.redaction"], limit: 5000);
+        var limitedTimelineRooms = sync.Rooms?.Join?
+            .Where(x => x.Value.Timeline?.Limited ?? false)
+            .Select(async x => {
+                var (roomId, roomData) = x;
+                var room = hs.GetRoom(roomId);
+                if (roomData.Timeline?.Limited == true) {
+                    msgQueue.EnqueueMessageAsync(filterConfigService.LogRoom, new MessageBuilder("m.notice")
+                        .WithColoredBody("FF0000", $"Room {roomId} has limited timeline, fetching! The room may be getting spammed?")
+                        .Build());
+                    roomData.Timeline.Events ??= [];
+                    var newEvents = await room.GetMessagesAsync(roomData.Timeline.PrevBatch ?? "", 500, filter: timelineFilter.ToJson(ignoreNull: true, indent: false));
+                    roomData.Timeline.Events.MergeBy(newEvents.Chunk, (x, y) => x.EventId == y.EventId, (x, y) => { });
+                }
+            })
+            .ToList();
+
+        if (limitedTimelineRooms?.Count > 0)
+            await Task.WhenAll(limitedTimelineRooms);
+
+        var tasks = Parallel.ForEachAsync(filters, async (filter, ct) => {
+            try {
+                Console.WriteLine("Processing filter {0}", filter.GetType().FullName);
+                await filter.ProcessSyncAsync(sync);
+            }
+            catch (Exception e) {
+                logger.LogError(e, "Error processing sync with filter {filter}", filter.GetType().FullName);
+                msgQueue.EnqueueMessageAsync(filterConfigService.LogRoom, new MessageBuilder("m.notice")
+                    .WithBody($"Error processing sync with filter {filter.GetType().FullName}: {e.Message}").Build());
+            }
+        });
+
+        await tasks;
+    }
+}
\ No newline at end of file