MCPcopy Create free account
hub / github.com/Mnexa-AI/e2a / createInboundMessage

Function createInboundMessage

internal/identity/store.go:1201–1262  ·  view source on GitHub ↗
(ctx context.Context, exec messageExecutor, id, agentID, senderEmail, recipient, emailMessageID, subject, conversationID, deliveryStatus string, rawMessage []byte, authHeaders map[string]string, authVerdict []byte, flagged bool, flagReason string, toRecipients, cc, replyTo []string, screening InboundScreening)

Source from the content-addressed store, hash-verified

1199}
1200
1201func createInboundMessage(ctx context.Context, exec messageExecutor, id, agentID, senderEmail, recipient, emailMessageID, subject, conversationID, deliveryStatus string, rawMessage []byte, authHeaders map[string]string, authVerdict []byte, flagged bool, flagReason string, toRecipients, cc, replyTo []string, screening InboundScreening) (*Message, error) {
1202 if id == "" {
1203 id = NewMessageID()
1204 }
1205 now := time.Now()
1206
1207 var authHeadersJSON []byte
1208 if authHeaders != nil {
1209 var err error
1210 authHeadersJSON, err = json.Marshal(authHeaders)
1211 if err != nil {
1212 return nil, fmt.Errorf("marshal auth headers: %w", err)
1213 }
1214 }
1215
1216 // Held messages (review/block) carry a review-queue status; everything else is
1217 // 'sent' (the inbound default — delivered).
1218 status := MessageStatusSent
1219 if screening.Status != "" {
1220 status = screening.Status
1221 }
1222
1223 m := &Message{
1224 ID: id,
1225 AgentID: agentID,
1226 Direction: "inbound",
1227 Sender: senderEmail,
1228 Recipient: recipient,
1229 ToRecipients: toRecipients,
1230 CC: cc,
1231 ReplyTo: replyTo,
1232 Subject: subject,
1233 EmailMessageID: emailMessageID,
1234 RawMessage: rawMessage,
1235 AuthHeaders: authHeaders,
1236 ConversationID: conversationID,
1237 DeliveryStatus: deliveryStatus,
1238 Flagged: flagged,
1239 FlagReason: flagReason,
1240 ReviewReason: screening.ReviewReason,
1241 ScanScore: screening.ScanScore,
1242 ScanAction: screening.ScanAction,
1243 Status: status,
1244 ApprovalExpiresAt: screening.ApprovalExpiresAt,
1245 CreatedAt: now,
1246 ExpiresAt: now.Add(MessageTTL),
1247 }
1248 // inbox_status column has CHECK constraint: must be 'unread', 'read', or NULL
1249 var inboxStatus *string
1250 if m.DeliveryStatus == "unread" || m.DeliveryStatus == "read" {
1251 inboxStatus = &m.DeliveryStatus
1252 }
1253 _, err := exec.Exec(ctx,
1254 `INSERT INTO messages (id, agent_id, direction, sender, recipient, to_recipients, cc, reply_to, subject, email_message_id, raw_message, auth_headers, auth_verdict, flagged, flag_reason, conversation_id, inbox_status, created_at, expires_at, review_reason, scan_score, scan_action, status, approval_expires_at)
1255 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24)`,
1256 m.ID, m.AgentID, m.Direction, m.Sender, m.Recipient, m.ToRecipients, m.CC, m.ReplyTo, m.Subject, m.EmailMessageID, m.RawMessage, authHeadersJSON, nullIfEmptyBytes(authVerdict), m.Flagged, nullIfEmptyString(m.FlagReason), m.ConversationID, inboxStatus, m.CreatedAt, m.ExpiresAt, nullIfEmptyString(m.ReviewReason), m.ScanScore, nullIfEmptyString(m.ScanAction), m.Status, m.ApprovalExpiresAt,
1257 )
1258 if err != nil {

Callers 2

CreateInboundMessageMethod · 0.85

Calls 4

NewMessageIDFunction · 0.85
nullIfEmptyBytesFunction · 0.85
nullIfEmptyStringFunction · 0.85
ExecMethod · 0.65

Tested by

no test coverage detected