Communication — Luồng xử lý trong code

Ánh xạ pipeline gửi tới worker & file trong Tux.

Các worker

Worker	Vai trò
Tux.Workers.Notification.Scheduler	Quét NotificationScheduler (và MessagingJob) Ready; đánh giá Notification Settings; đẩy việc sang Generator.
Tux.Workers.Notification.Generator	Render template, resolve recipients; sinh UserMessage và/hoặc OutgoingMessage.
Tux.Workers.Notification.Distributor	"Claim" rồi gửi thật qua provider (SendGrid/Twilio).
Webhook & Tracker	Nhận callback provider, cập nhật trạng thái OutgoingMessage.
Worker Automation Scheduler	Quét AutomationScheduler (reminder/report), đẩy việc giao hàng sang NotificationScheduler.

Generator: route theo loại notification

Generator dựa trên loại để chọn handler render. Ví dụ các handler id-only (đọc EventData là entity id):

// Trong Generator
var id = JsonConvert.DeserializeObject<int>((string)job.EventData);
// route: IncidentSubmitted -> HandleNotificationAsync -> HandleIncidentSubmittedAsync
//        -> NotificationService.SendIncidentSubmittedNotification(...)

Handler mẫu: HandleAccountNewRegistration_EmployeeAsync, HandleStaffProfileUpdateAsync, HandleIncidentSubmittedAsync. Mỗi handler resolve recipients (từ NotificationSetting subscriber + dữ liệu job), render template, sinh OutgoingMessage.

Distributor: claim trước khi gửi

// Pseudo: cập nhật có điều kiện
// UPDATE OutgoingMessage SET Status = Sending
//   WHERE PartitionKey=@pk AND RowKey=@rk AND Status IN (Created, RetryReady)
// Nếu 0 dòng bị đổi → tin đã/đang gửi → thoát, KHÔNG gọi provider

Sau khi gửi thành công → SentToProvider + lưu provider message id. Webhook đẩy tiếp tới Delivered/Opened.

Campaign

MessagingJob (SQL) điều phối payload campaign/report; MessageJobBlobName trỏ container messaging-job (chỉ tên/đường dẫn — API authorize trước khi lộ nội dung). Worker Notification Scheduler query cả NotificationScheduler và MessagingJob. Retry tái dùng blob payload tương thích; không tương thích → dừng gửi.

UserMessage (in-app)

Ghi dữ liệu

CustomerInboxPersistService resolve recipient trước khi ghi: đúng một owner hợp lệ mới được persist; recipient staff còn chịu rollout flag/allowlist. CampaignInAppUserMessagePersistence ghi trực tiếp canonical owner partition; Incident cũ đi qua LegacyIncidentUserMessagePersistence và có dual-read legacy key khi chuyển sang Account_{AccountId}.

1. Chuẩn bị body HTML/text trong Blob container message-content; row chỉ giữ MessageBodyBlobName. Body có thể dùng blob đã có hoặc tên content-addressed sha256-{hash} tùy producer.
2. Tạo metadata UserMessageEntity: recipient/source scope, Title, MessagePreview, action, display labels, entity coordinates, StatusId, timestamps và CompatibilityHash.
3. EnsureUserMessageAsync dùng Add insert-if-absent. Retry đọc lại row và giữ recipient-owned state; immutable mismatch trả VERSION_CONFLICT. Chỉ row tạo mới mới publish realtime event.

Khóa lưu trữ

Canonical PartitionKey:
  Account_{AccountId}
  Employee_{EmployeeId}
  EnterpriseUser_{EnterpriseUserId}

Legacy compatibility PartitionKey:
  Customer_{UserId}

Common deterministic RowKey examples:
  inbox_{sourcePartitionKey}_{deterministicHash}
  {reverseTicks}-{deterministicRowKey}
  legacy Incident: {reverseTicks}

Customer_{UserId} là compatibility-read/dedup cho dữ liệu cũ; delivery mới không dựng partition này. RowKey là identity theo producer, không dùng UserId đơn độc để authorize. reverseTicks phải đến từ timestamp nguồn ổn định, không phải write-time.

Đọc và truy vấn

UserMessageInboxController gọi UserMessageService, service resolve authenticated portal → owner scopes trước khi chạm Table/Blob:

• List: GET /api/UserMessageInbox?take&cursor&messageCategoryId&globalView; query từng PartitionKey, chỉ lấy Unread/Read, filter category/Account/BU/Site ở server.
• Paging: mặc định 20, tối đa 100; Azure continuation token của từng scope nằm trong NextCursor opaque. Multi-scope merge lấy head mới nhất rồi chỉ advance scope vừa chọn.
• Unread: GET /api/UserMessageInbox/UnreadCount; đọc _InboxUnread counter trong cùng Table, khởi tạo bằng bounded scan nếu counter chưa có. Contract active trả một tổng unread.
• Detail/body: point-read bằng PartitionKey + RowKey; body gọi Blob sau authorization, không trả permanent Blob URL và list chỉ trả preview.
• State: PUT .../{action} với ETag hiện tại; MarkRead, Archive, Save, Unsave là conditional update. Save set SavedOn, Unsave clear SavedOn, không copy row và không reset status khác.

Retention job tìm unsaved row quá ba tháng qua maintenance index _InboxUserMessageMaintenance_{yyyyMM} rồi xóa conditional theo ETag; saved row được miễn.

Realtime Inbox

Thành phần	Vị trí	Vai trò
Publisher	Tux.Service.AzureService.RedisUserMessageRealtimePublisher	Ghi một inbox.changed vào Redis Stream sau khi tạo UserMessage mới.
Stream consumer	Tux.Api.Infrastructure.Inbox.UserMessageRealtimeStreamConsumer	Mỗi API instance đọc và ACK event, rồi fanout tới connection cục bộ.
Connection manager	Tux.Api.Infrastructure.Inbox.UserMessageRealtimeConnectionManager	Đăng ký theo JWT subject, replay Last-Event-ID, giới hạn buffer và yêu cầu sync khi cần.
SSE endpoint	Tux.Api.Controllers.UserMessageInboxController.StreamAsync	Trả text/event-stream, heartbeat và event metadata; không trả body Inbox.
Tuke client	Tuke/app/lib/user-message-realtime.ts + Tuke/app/stores/userMessage.ts	Parse event, reconnect/backoff, dedupe và refresh REST unread/first page.
Topi bridge	postMessage từ Tuke	Cập nhật badge host; Topi không mở thêm SSE connection cho iframe.

Luồng xử lý là:

1. Producer persist UserMessage qua persistence boundary hiện có.
2. Chỉ outcome tạo mới phát event; re-ensure hoặc state update không phát creation event.
3. Tux API tiêu thụ Redis Stream và gửi event tới đúng các connection của JWT sub.
4. Tuke dùng event làm invalidation, sau đó gọi REST để lấy dữ liệu theo scope hiện tại.
5. Reconnect dùng replay; nếu event đã hết retention thì nhận inbox.sync-required và revalidate REST.

Ví dụ thực tế: Tux worker tạo một UserMessage cho nhân viên. → Publisher ghi event vào Redis; Tuke nhận được event dù connection đang ở API instance khác. → Tuke refresh unread, còn quyền đọc vẫn do REST.

Notification Settings điều khiển gì

Bước Scheduler đọc cấu hình theo BU:

• NotificationOption — danh mục loại notification (phải seed loại mới vào đây).
• NotificationSetting — bật/tắt + subscriber theo từng BU.

Loại chưa seed hoặc không có subscriber → handler skip gracefully (không gửi).

DeliveryDispatchResult

Kết quả mỗi lần dispatch lưu ở AutomationScheduler.DispatchResultJson: delivery path + tọa độ (notification/scheduler/messaging-job/user-message/outgoing-message) + SkippedReasonCode/ConflictReasonCode/Summary. Nhỏ gọn, merge khi retry, không bao giờ bị xóa.

Test trọng yếu

• Direct reminder retry → một NotificationJob; Generator retry → một OutgoingMessage per recipient/method; tin Distributor trùng → không gọi provider lần hai.
• UI-only reminder → UserMessage, không OutgoingMessage; UserMessage retry với source ticks ổn định → một dòng.
• Scheduler thiếu được tạo lại qua dispatch guard; hash không khớp → dừng gửi.
• Storage adapters giữ nguyên khóa do caller cấp.
• Report/campaign retry tái dùng MessagingJob + Blob tương thích.

Traceability và report run

• Generator resolve source trước khi build OutgoingMessage và dùng SourceKey/partition helper; invoice dùng invoice GUID, account-scoped booking/statement dùng account GUID, source không resolve thì fallback.
• Distributor lease có ClaimId/thời gian claim; claim stale được recovery qua delayed check, còn provider progress không bị retry ghi đè.
• Report Subscription dùng ReportSubscriptionScheduler làm run history, lưu file trước khi gửi; run quá lớn chuyển link-only thay vì fail do attachment.
• Notification Setting list đọc aggregate theo BU, ghi toggle Site/BU contact bằng batch path; detail page không còn sở hữu toggle copy.

Ví dụ thực tế: Generator retry sau khi process đã tạo file report. → Không tạo file/delivery đôi. → Reuse scheduler/run coordinate và outgoing claim; history giữ trạng thái run trước đó.

Tiếp theo: Case thường gặp.