V15 / 313404a015e8
BridgeStore.cs
Services/MessageBridge/BridgeStore.cs
Unchanged from the preceding versionCS
using System.Security.Cryptography;
using System.Text;
using System.Text.Json;
using System.Text.RegularExpressions;
namespace Lilith.MessageBridge;
public sealed record ChannelMessage(string Id, string Room, string Sender, string Text, string Direction, DateTimeOffset AtUtc, string Status = "queued");
public sealed record Proposal(string Id, string Room, string Text, string ReplyTo, string Key, string Signature, string Status = "prepared");
public sealed record ChannelState(string Nickname, string Adapter, List<ChannelMessage> Messages, List<Proposal> Proposals);
public sealed record BridgeRequest(string Nickname = "", string MessageId = "", string Room = "", string Text = "", string ReplyTo = "", string Key = "", string ProposalId = "", string Signature = "");
/// <summary>One bridge process owns durable, atomic local dummy channel state.</summary>
public sealed class BridgeStore : IDisposable
{
private readonly string _root;
private readonly FileStream _lease;
private readonly SemaphoreSlim _gate = new(1, 1);
public BridgeStore(string root)
{
_root = Path.GetFullPath(root); Directory.CreateDirectory(_root);
_lease = new(Path.Combine(_root, ".lease"), FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.None);
}
public void Dispose() { _lease.Dispose(); _gate.Dispose(); }
private string PathFor(string nickname)
{
if (!Regex.IsMatch(nickname, "^[a-z][a-z0-9_-]{0,47}$")) throw new ArgumentException("Invalid nickname.");
return Path.Combine(_root, nickname + ".json");
}
private ChannelState Read(string nickname) => JsonSerializer.Deserialize<ChannelState>(File.ReadAllText(PathFor(nickname))) ?? throw new InvalidDataException("Invalid channel state.");
private void Save(ChannelState state)
{
string path = PathFor(state.Nickname), temp = path + ".tmp";
File.WriteAllText(temp, JsonSerializer.Serialize(state)); File.Move(temp, path, true);
}
private static string Required(string text, int maximum)
=> !string.IsNullOrWhiteSpace(text) && text.Length <= maximum && !text.Any(c => char.IsControl(c) && c != '\n' && c != '\t') ? text : throw new ArgumentException("Invalid message field.");
public async Task AddDummyAsync(string nickname, CancellationToken ct = default)
{
await _gate.WaitAsync(ct); try
{
if (File.Exists(PathFor(nickname))) throw new InvalidOperationException("Channel already exists.");
Save(new(nickname, "dummy-discord", [], []));
} finally { _gate.Release(); }
}
public async Task<string> ArriveAsync(string nickname, string room, string sender, string text, CancellationToken ct = default)
{
await _gate.WaitAsync(ct); try
{
var state = Read(nickname);
if (state.Messages.Count >= 1000) throw new InvalidOperationException("Dummy channel retention limit reached; create another isolated test channel.");
var message = new ChannelMessage(Guid.NewGuid().ToString("N"), Required(room, 100), Required(sender, 100), Required(text, 4000), "incoming", DateTimeOffset.UtcNow);
state.Messages.Add(message); Save(state); return message.Id;
} finally { _gate.Release(); }
}
public async Task<object> ExecuteAsync(string action, BridgeRequest request, CancellationToken ct)
{
await _gate.WaitAsync(ct); try
{
if (action == "channels") return new { channels = Directory.EnumerateFiles(_root, "*.json").OrderBy(p => p, StringComparer.Ordinal).Where(p => string.CompareOrdinal(Path.GetFileNameWithoutExtension(p), request.Key) > 0).Take(20).Select(p => Read(Path.GetFileNameWithoutExtension(p))).Select(s => new { s.Nickname, s.Adapter, source = "dummy" }).ToArray() };
var state = Read(request.Nickname);
switch (action)
{
case "inbox": return new { source = "dummy", state.Nickname, messages = state.Messages.Where(m => m.Direction == "incoming" && m.Status == "queued").Take(10).ToArray(), moreAvailable = state.Messages.Count(m => m.Direction == "incoming" && m.Status == "queued") > 10 };
case "status": return new { source = "dummy", state.Nickname, messages = state.Messages.TakeLast(20).ToArray(), proposals = state.Proposals.Where(p => p.Status == "prepared").Take(20).ToArray() };
case "proposal": return state.Proposals.Single(p => p.Id == request.ProposalId);
case "claim":
var found = state.Messages.Single(m => m.Id == request.MessageId && m.Direction == "incoming");
if (found.Status != "queued") return new { status = "already_handled", source = "dummy" };
state.Messages[state.Messages.IndexOf(found)] = found with { Status = "claimed" }; Save(state);
return new { status = "claimed", source = "dummy", message = found };
case "prepare":
Required(request.Key, 100); Required(request.Room, 100); Required(request.Text, 2000);
if (request.Text.Contains("@everyone", StringComparison.OrdinalIgnoreCase) || request.Text.Contains("@here", StringComparison.OrdinalIgnoreCase) || request.Text.Contains("<@", StringComparison.Ordinal)) throw new ArgumentException("Mass mentions and user/role mentions are unsupported.");
var previous = state.Proposals.FirstOrDefault(p => p.Key == request.Key);
if (previous is not null)
{
if (previous.Room != request.Room || previous.Text != request.Text || previous.ReplyTo != request.ReplyTo) throw new InvalidOperationException("Idempotency key payload differs.");
return previous;
}
if (state.Proposals.Count >= 1000) throw new InvalidOperationException("Dummy proposal retention limit reached.");
if (request.ReplyTo.Length > 0)
{
var origin = state.Messages.Single(m => m.Id == request.ReplyTo && m.Direction == "incoming" && m.Room == request.Room);
if (origin.Status != "claimed" || state.Proposals.Any(p => p.ReplyTo == origin.Id)) throw new InvalidOperationException("Incoming message is not available for a reply.");
}
string signature = Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(JsonSerializer.Serialize(new { request.Nickname, request.Room, request.Text, request.ReplyTo }))));
var proposal = new Proposal(Guid.NewGuid().ToString("N"), request.Room, request.Text, request.ReplyTo, request.Key, signature);
state.Proposals.Add(proposal); Save(state); return proposal;
case "send":
case "reject":
var pending = state.Proposals.Single(p => p.Id == request.ProposalId);
if (pending.Signature != request.Signature) throw new InvalidOperationException("Proposal signature differs.");
if (pending.Status is "simulated_sent" or "rejected") return new { status = pending.Status, source = "dummy", state.Nickname, proposal = pending };
string status = action == "send" ? "simulated_sent" : "rejected";
if (action == "send") state.Messages.Add(new(Guid.NewGuid().ToString("N"), pending.Room, "Lilith", pending.Text, "outgoing", DateTimeOffset.UtcNow, status));
var updated = pending with { Status = status }; state.Proposals[state.Proposals.IndexOf(pending)] = updated;
if (pending.ReplyTo.Length > 0)
{
int index = state.Messages.FindIndex(m => m.Id == pending.ReplyTo);
state.Messages[index] = state.Messages[index] with { Status = status };
}
Save(state); return new { status, source = "dummy", state.Nickname, proposal = updated };
default: throw new ArgumentException("Unknown bridge operation.");
}
} finally { _gate.Release(); }
}
}