Forge
search
csharp4405de34
1using System.Text.Json;
2using System.Text.Json.Nodes;
3using IntercomWire;
4using IntercomService.Contracts;
5using IntercomService.Data;
6using Microsoft.EntityFrameworkCore;
7
8namespace IntercomService.Services;
9
10public sealed class TransportEventService(IntercomDbContext db, SseEventHub sse, TeamMembershipService teams)
11{
12 public async Task<TopicEntity> EnsureGeneralTopicAsync(string teamId, CancellationToken ct)
13 {
14 var existing = await db.Topics
15 .Where(x => x.TeamId == teamId && x.Title == "general")
16 .FirstOrDefaultAsync(ct)
17 .ConfigureAwait(false);
18
19 if (existing is not null)
20 return existing;
21
22 var topic = new TopicEntity
23 {
24 TopicId = Guid.NewGuid().ToString("N"),
25 TeamId = teamId,
26 Title = "general",
27 SpineKey = "general",
28 CreatedAtUtc = DateTimeOffset.UtcNow,
29 };
30 db.Topics.Add(topic);
31 await db.SaveChangesAsync(ct).ConfigureAwait(false);
32 return topic;
33 }
34
35 public async Task<TopicEntity> EnsureTopicBySpineAsync(
36 string teamId,
37 string spineKey,
38 string title,
39 CancellationToken ct)
40 {
41 var key = spineKey.Trim();
42 var existing = await db.Topics
43 .AsNoTracking()
44 .FirstOrDefaultAsync(x => x.TeamId == teamId && x.SpineKey == key, ct)
45 .ConfigureAwait(false);
46 if (existing is not null)
47 return existing;
48
49 var topic = new TopicEntity
50 {
51 TopicId = Guid.NewGuid().ToString("N"),
52 TeamId = teamId,
53 Title = string.IsNullOrWhiteSpace(title) ? key : title.Trim(),
54 SpineKey = key,
55 CreatedAtUtc = DateTimeOffset.UtcNow,
56 };
57 db.Topics.Add(topic);
58 await db.SaveChangesAsync(ct).ConfigureAwait(false);
59 return topic;
60 }
61
62 public async Task<(TransportEventEnvelopeDto? Envelope, string? Error)> AppendAsync(
63 string teamId,
64 string topicId,
65 AppendEventRequest request,
66 string operatorMemberId,
67 string operatorDisplayName,
68 string clientKind,
69 CancellationToken ct)
70 {
71 if (string.IsNullOrWhiteSpace(request.ClientEventId))
72 return (null, "client_event_id required");
73
74 if (!IntercomWireTransportEventKinds.SyncDefault.Contains(request.EventKind))
75 return (null, $"event_kind '{request.EventKind}' not allowed");
76
77 var operatorRole = await teams.GetTeamRoleAsync(teamId, operatorMemberId, ct).ConfigureAwait(false);
78 if (operatorRole is null)
79 return (null, "not_a_member");
80
81 var role = ResolveSenderRole(request);
82
83 var topic = await db.Topics.FirstOrDefaultAsync(x => x.TopicId == topicId && x.TeamId == teamId, ct)
84 .ConfigureAwait(false);
85 if (topic is null)
86 return (null, "topic not found");
87
88 string senderMemberId;
89 string senderDisplayName;
90 string payloadJson;
91
92 if (string.Equals(role, "agent", StringComparison.Ordinal))
93 {
94 if (!TeamRoleAuthorization.CanPublishMessages(operatorRole))
95 return (null, "guest_cannot_publish");
96
97 var agentMemberId = request.Sender?.MemberId?.Trim() ?? "";
98 if (string.IsNullOrWhiteSpace(agentMemberId))
99 return (null, "agent_member_id required");
100
101 var agent = await db.TeamMembers
102 .Include(x => x.Member)
103 .AsNoTracking()
104 .FirstOrDefaultAsync(
105 x => x.TeamId == teamId && x.MemberId == agentMemberId,
106 ct)
107 .ConfigureAwait(false);
108
109 if (agent?.Member is null
110 || !string.Equals(agent.Member.MemberKind, MemberKinds.Agent, StringComparison.Ordinal))
111 return (null, "agent_not_in_team");
112
113 senderMemberId = agentMemberId;
114 senderDisplayName = string.IsNullOrWhiteSpace(request.Sender?.DisplayName)
115 ? agent.Member.DisplayName
116 : request.Sender.DisplayName.Trim();
117 payloadJson = InjectOperatorMemberId(request.Payload, operatorMemberId);
118 }
119 else
120 {
121 if (!TeamRoleAuthorization.CanPublishMessages(operatorRole))
122 return (null, "guest_cannot_publish");
123
124 senderMemberId = operatorMemberId;
125 senderDisplayName = operatorDisplayName;
126 payloadJson = request.Payload.GetRawText();
127 }
128
129 var dup = await db.TransportEvents
130 .AsNoTracking()
131 .FirstOrDefaultAsync(x => x.TeamId == teamId && x.ClientEventId == request.ClientEventId, ct)
132 .ConfigureAwait(false);
133 if (dup is not null)
134 return (ToEnvelope(dup), null);
135
136 var maxSeq = await db.TransportEvents
137 .Where(x => x.TeamId == teamId)
138 .Select(x => (long?)x.Seq)
139 .MaxAsync(ct)
140 .ConfigureAwait(false) ?? 0;
141
142 var occurred = DateTimeOffset.TryParse(request.OccurredAtUtc, out var at)
143 ? at
144 : DateTimeOffset.UtcNow;
145
146 var entity = new TransportEventEntity
147 {
148 TeamId = teamId,
149 Seq = maxSeq + 1,
150 TopicId = topicId,
151 ClientEventId = request.ClientEventId.Trim(),
152 EventKind = request.EventKind,
153 SenderMemberId = senderMemberId,
154 SenderDisplayName = senderDisplayName,
155 SenderRole = role,
156 ClientKind = request.Sender?.ClientKind ?? clientKind,
157 PayloadJson = payloadJson,
158 OccurredAtUtc = occurred,
159 };
160
161 db.TransportEvents.Add(entity);
162 await db.SaveChangesAsync(ct).ConfigureAwait(false);
163
164 var envelope = ToEnvelope(entity);
165 sse.Publish(teamId, envelope);
166 return (envelope, null);
167 }
168
169 public async Task<IReadOnlyList<TransportEventEnvelopeDto>> ListAsync(
170 string teamId,
171 string topicId,
172 long? afterSeq,
173 int limit,
174 CancellationToken ct)
175 {
176 limit = Math.Clamp(limit, 1, 500);
177 var query = db.TransportEvents.AsNoTracking()
178 .Where(x => x.TeamId == teamId && x.TopicId == topicId);
179
180 if (afterSeq is > 0)
181 query = query.Where(x => x.Seq > afterSeq.Value);
182
183 var rows = await query
184 .OrderBy(x => x.Seq)
185 .Take(limit)
186 .ToListAsync(ct)
187 .ConfigureAwait(false);
188
189 return rows.ConvertAll(ToEnvelope);
190 }
191
192 public async Task<IReadOnlyList<TransportEventEnvelopeDto>> ListTeamAsync(
193 string teamId,
194 long? afterSeq,
195 int limit,
196 CancellationToken ct)
197 {
198 limit = Math.Clamp(limit, 1, 500);
199 var query = db.TransportEvents.AsNoTracking().Where(x => x.TeamId == teamId);
200 if (afterSeq is > 0)
201 query = query.Where(x => x.Seq > afterSeq.Value);
202
203 var rows = await query
204 .OrderBy(x => x.Seq)
205 .Take(limit)
206 .ToListAsync(ct)
207 .ConfigureAwait(false);
208
209 return rows.ConvertAll(ToEnvelope);
210 }
211
212 private static string ResolveSenderRole(AppendEventRequest request)
213 {
214 if (!string.IsNullOrWhiteSpace(request.Sender?.SenderRole))
215 return request.Sender.SenderRole.Trim();
216
217 return InferSenderRoleFromPayload(request) ?? "human";
218 }
219
220 private static string? InferSenderRoleFromPayload(AppendEventRequest request)
221 {
222 if (!request.EventKind.StartsWith("message_", StringComparison.Ordinal))
223 return null;
224
225 try
226 {
227 if (!request.Payload.TryGetProperty("role", out var roleEl))
228 return null;
229 var role = roleEl.GetString();
230 return role?.ToLowerInvariant() switch
231 {
232 "assistant" => "agent",
233 "system" => "system",
234 _ => "human",
235 };
236 }
237 catch
238 {
239 return null;
240 }
241 }
242
243 private static string InjectOperatorMemberId(JsonElement payload, string operatorMemberId)
244 {
245 var root = JsonNode.Parse(payload.GetRawText()) as JsonObject ?? new JsonObject();
246 root["operator_member_id"] = operatorMemberId;
247 return root.ToJsonString();
248 }
249
250 public static TransportEventEnvelopeDto ToEnvelope(TransportEventEntity e) =>
251 new(
252 SchemaVersion: 1,
253 Seq: e.Seq,
254 TeamId: e.TeamId,
255 TopicId: e.TopicId,
256 ClientEventId: e.ClientEventId,
257 OccurredAtUtc: e.OccurredAtUtc.ToString("O"),
258 EventKind: e.EventKind,
259 Sender: new SenderDto(e.SenderMemberId, e.SenderDisplayName, e.SenderRole, e.ClientKind),
260 Payload: JsonDocument.Parse(e.PayloadJson).RootElement.Clone());
261}
262
View only · write via MCP/CIDE