feat/fix: Real-Time Encrypted Chat Engine, Redis Streams Event Bus, and Chatbot Backend

services/finance/internal/delivery/grpc/interceptors.go

@@ -65,18 +65,35 @@ func RequestIDInterceptor() grpc.UnaryServerInterceptor {
 	}
 }
 
-// TimeoutInterceptor enforces request timeout.
+// longRunningMethods maps gRPC method suffixes to extended timeouts for RPCs
+// that routinely exceed the default 60s (batch compute, bulk import/export).
+var longRunningMethods = map[string]time.Duration{
+	"TriggerMbBatch":              5 * time.Minute,
+	"ExecutePushToHead":           5 * time.Minute,
+	"ProcessChunkInternal":        5 * time.Minute,
+	"ImportCostApplicableParams":  3 * time.Minute,
+	"ImportCostProductParameters": 3 * time.Minute,
+}
+
+// TimeoutInterceptor enforces request timeout. RPCs listed in
+// longRunningMethods get an extended deadline; all others use the default.
 func TimeoutInterceptor(timeout time.Duration) grpc.UnaryServerInterceptor {
 	return func(
 		ctx context.Context,
 		req interface{},
-		_ *grpc.UnaryServerInfo,
+		info *grpc.UnaryServerInfo,
 		handler grpc.UnaryHandler,
 	) (interface{}, error) {
-		// Check if context already has deadline
 		if _, ok := ctx.Deadline(); !ok {
+			effective := timeout
+			for suffix, dur := range longRunningMethods {
+				if strings.HasSuffix(info.FullMethod, "/"+suffix) {
+					effective = dur
+					break
+				}
+			}
 			var cancel context.CancelFunc
-			ctx, cancel = context.WithTimeout(ctx, timeout)
+			ctx, cancel = context.WithTimeout(ctx, effective)
 			defer cancel()
 		}
 

services/iam/cmd/server/main.go

@@ -11,12 +11,17 @@ import (
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 
+	"encoding/hex"
+
 	iamv1 "github.com/mutugading/goapps-backend/gen/iam/v1"
 	authapp "github.com/mutugading/goapps-backend/services/iam/internal/application/auth"
+	appChat "github.com/mutugading/goapps-backend/services/iam/internal/application/chat"
 	appnotif "github.com/mutugading/goapps-backend/services/iam/internal/application/notification"
 	grpcdelivery "github.com/mutugading/goapps-backend/services/iam/internal/delivery/grpc"
 	httpdelivery "github.com/mutugading/goapps-backend/services/iam/internal/delivery/httpdelivery"
+	chatinfra "github.com/mutugading/goapps-backend/services/iam/internal/infrastructure/chat"
 	"github.com/mutugading/goapps-backend/services/iam/internal/infrastructure/config"
+	iamcrypto "github.com/mutugading/goapps-backend/services/iam/internal/infrastructure/crypto"
 	emailinfra "github.com/mutugading/goapps-backend/services/iam/internal/infrastructure/email"
 	"github.com/mutugading/goapps-backend/services/iam/internal/infrastructure/jwt"
 	notifinfra "github.com/mutugading/goapps-backend/services/iam/internal/infrastructure/notification"
@@ -204,6 +209,16 @@ func run() error {
 		notifStream, validationHelper,
 	).WithRequestHandler(notifRequestHandler)
 
+	// ── Chat + Presence ────────────────────────────────────────────────────
+	chatHandler, presenceGRPCHandler := initChatServices(chatDeps{
+		masterKeyHex:    cfg.Chat.MasterKey,
+		db:              db,
+		redisClient:     redisClient,
+		notifCreate:     notifCreate,
+		emailDispatcher: notifEmailDispatcher,
+		storageSvc:      storageSvc,
+	})
+
 	// Setup gRPC server with interceptor chain (pass JWT + session cache + session repo for auth & activity tracking)
 	grpcServer, err := grpcdelivery.NewServer(&cfg.Server, db, jwtService, sessionCache, sessionRepo, cfg.Security.InternalServiceToken)
 	if err != nil {
@@ -232,6 +247,12 @@ func run() error {
 	iamv1.RegisterNotificationServiceServer(gs, notificationHandler)
 	iamv1.RegisterWorkflowTemplateServiceServer(gs, workflowTemplateHandler)
 	iamv1.RegisterWorkflowInstanceServiceServer(gs, workflowInstanceHandler)
+	if chatHandler != nil {
+		iamv1.RegisterChatServiceServer(gs, chatHandler)
+	}
+	if presenceGRPCHandler != nil {
+		iamv1.RegisterPresenceServiceServer(gs, presenceGRPCHandler)
+	}
 
 	// Start gRPC server
 	go func() {
@@ -273,6 +294,80 @@ func run() error {
 	return nil
 }
 
+// chatDeps bundles the dependencies needed to wire the chat + presence stack.
+type chatDeps struct {
+	masterKeyHex    string
+	db              *postgres.DB
+	redisClient     *redisinfra.Client
+	notifCreate     *appnotif.CreateHandler
+	emailDispatcher appnotif.EmailDispatcher
+	storageSvc      storageinfra.Service
+}
+
+// initChatServices builds the chat and presence gRPC handlers. Both are nil
+// when CHAT_MASTER_KEY is unset (chat disabled); presence is nil when Redis is
+// unavailable. Extracted from run to keep its cognitive complexity in bounds.
+func initChatServices(d chatDeps) (*grpcdelivery.ChatHandler, *grpcdelivery.PresenceHandler) {
+	if d.masterKeyHex == "" {
+		log.Warn().Msg("CHAT_MASTER_KEY not set — chat services disabled")
+		return nil, nil
+	}
+	chatMasterKey, hexErr := hex.DecodeString(d.masterKeyHex)
+	if hexErr != nil || len(chatMasterKey) != 32 {
+		log.Fatal().Msg("CHAT_MASTER_KEY must be a 64-char hex string (32 bytes)")
+	}
+	chatEnc, encErr := iamcrypto.NewEncryptor(chatMasterKey)
+	if encErr != nil {
+		log.Fatal().Err(encErr).Msg("Failed to init chat encryptor")
+	}
+
+	chatBroadcaster := chatinfra.NewBroadcaster()
+	var presenceSvc *chatinfra.PresenceService
+	if d.redisClient != nil {
+		chatBroadcaster = chatinfra.NewRedisBroadcaster(d.redisClient.Client)
+		presenceSvc = chatinfra.NewPresenceService(d.redisClient.Client)
+	}
+
+	convRepo := postgres.NewChatConversationRepository(d.db)
+	msgRepo := postgres.NewChatMessageRepository(d.db)
+	receiptRepo := postgres.NewChatReadReceiptRepository(d.db)
+	attRepo := postgres.NewChatAttachmentRepository(d.db)
+	userResolver := postgres.NewChatUserResolver(d.db)
+
+	sendMsgHandler := appChat.NewSendMessageHandler(convRepo, msgRepo, receiptRepo, chatEnc, chatBroadcaster)
+	sendMsgHandler.WithAttachments(attRepo)
+	if presenceSvc != nil && d.redisClient != nil {
+		sendMsgHandler.WithOfflineNotification(presenceSvc, d.notifCreate, d.emailDispatcher, d.redisClient.Client, userResolver)
+	}
+
+	chatHandler := grpcdelivery.NewChatHandler(
+		appChat.NewCreateDirectHandler(convRepo, chatEnc),
+		appChat.NewCreateGroupHandler(convRepo, chatEnc),
+		appChat.NewGetConversationHandler(convRepo),
+		appChat.NewListConversationsHandler(convRepo, msgRepo, chatEnc),
+		appChat.NewLeaveConversationHandler(convRepo),
+		sendMsgHandler,
+		appChat.NewEditMessageHandler(convRepo, msgRepo, chatEnc, chatBroadcaster, userResolver),
+		appChat.NewDeleteMessageHandler(convRepo, msgRepo, chatBroadcaster),
+		appChat.NewListMessagesHandler(convRepo, msgRepo, chatEnc),
+		appChat.NewMarkReadHandler(convRepo, msgRepo, receiptRepo, chatBroadcaster),
+		appChat.NewSetTypingHandler(convRepo, presenceSvc, chatBroadcaster),
+		appChat.NewStreamHandler(chatBroadcaster),
+		appChat.NewGetEditHistoryHandler(convRepo, msgRepo, chatEnc),
+		appChat.NewClearHistoryHandler(convRepo),
+		appChat.NewUploadAttachmentHandler(convRepo, attRepo, d.storageSvc),
+		attRepo,
+		userResolver,
+	)
+
+	var presenceGRPCHandler *grpcdelivery.PresenceHandler
+	if presenceSvc != nil {
+		presenceGRPCHandler = grpcdelivery.NewPresenceHandler(presenceSvc)
+	}
+	log.Info().Msg("Chat + Presence services initialized")
+	return chatHandler, presenceGRPCHandler
+}
+
 // setupLogger configures the application logger.
 func setupLogger() {
 	zerolog.TimeFieldFormat = time.RFC3339

services/iam/config.yaml

@@ -106,6 +106,9 @@ storage:
   region: us-east-1
   public_url: ""                   # Override for external access URL, e.g., https://staging-goapps.mutugading.com/storage
 
+chat:
+  master_key: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"  # 64-char hex (32 bytes) — dev only, override via CHAT_MASTER_KEY env in staging/prod
+
 logging:
   level: debug  # debug, info, warn, error
   format: console  # console, json

services/iam/internal/application/chat/add_participants_handler.go

@@ -0,0 +1,45 @@
+// Package chat implements the application-layer use cases for chat
+// conversations, messages, attachments, presence, and real-time events.
+package chat
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/google/uuid"
+
+	"github.com/mutugading/goapps-backend/services/iam/internal/domain/chat"
+)
+
+// AddParticipantsHandler adds participants to a group conversation.
+type AddParticipantsHandler struct {
+	convRepo chat.ConversationRepository
+}
+
+// NewAddParticipantsHandler constructs the handler.
+func NewAddParticipantsHandler(convRepo chat.ConversationRepository) *AddParticipantsHandler {
+	return &AddParticipantsHandler{convRepo: convRepo}
+}
+
+// Handle adds userIDs to the conversation. Caller must be ADMIN or OWNER.
+func (h *AddParticipantsHandler) Handle(ctx context.Context, callerID, convID uuid.UUID, userIDs []uuid.UUID) error {
+	conv, err := h.convRepo.GetByID(ctx, convID)
+	if err != nil {
+		return err
+	}
+	p := conv.FindParticipant(callerID)
+	if p == nil || !p.IsActive() {
+		return fmt.Errorf("add participants: %w", chat.ErrNotParticipant)
+	}
+	if !p.Role().IsAdminOrOwner() {
+		return fmt.Errorf("add participants: %w", chat.ErrNotAdmin)
+	}
+	newParts := make([]*chat.Participant, 0, len(userIDs))
+	for _, uid := range userIDs {
+		if err := conv.AddParticipant(uid, chat.RoleMember); err != nil {
+			return fmt.Errorf("add participants: %w", err)
+		}
+		newParts = append(newParts, conv.FindParticipant(uid))
+	}
+	return h.convRepo.AddParticipants(ctx, convID, newParts)
+}

services/iam/internal/application/chat/broadcast_helpers.go

@@ -0,0 +1,104 @@
+package chat
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/rs/zerolog/log"
+
+	iamv1 "github.com/mutugading/goapps-backend/gen/iam/v1"
+	domainChat "github.com/mutugading/goapps-backend/services/iam/internal/domain/chat"
+	chatinfra "github.com/mutugading/goapps-backend/services/iam/internal/infrastructure/chat"
+	"github.com/mutugading/goapps-backend/services/iam/internal/infrastructure/postgres"
+)
+
+// decryptionErrorBody is the placeholder body shown when a message body
+// fails to decrypt (e.g. master key rotated out from under an old message).
+const decryptionErrorBody = "[decryption error]"
+
+// resolveSenderName looks up a display name for senderID via userResolver,
+// falling back to username, then an empty string if the resolver is nil or
+// the lookup fails.
+func resolveSenderName(ctx context.Context, userResolver *postgres.ChatUserResolver, senderID uuid.UUID) string {
+	if userResolver == nil {
+		return ""
+	}
+	infos, err := userResolver.ResolveUsers(ctx, []uuid.UUID{senderID})
+	if err != nil {
+		log.Warn().Err(err).Str("user", senderID.String()).Msg("chat: resolve sender name")
+		return ""
+	}
+	info := infos[senderID]
+	if info == nil {
+		return ""
+	}
+	if info.FullName != "" {
+		return info.FullName
+	}
+	return info.Username
+}
+
+// broadcastMessageEvent publishes a message-related event (created, edited) to
+// every active participant of conv via the given broadcaster.
+func broadcastMessageEvent(broadcaster *chatinfra.Broadcaster, conv *domainChat.Conversation, msg *domainChat.Message, plainBody, eventType, senderName string, attachments []*iamv1.AttachmentProto) {
+	msgProto := domainMsgToProto(msg, plainBody)
+	msgProto.SenderName = senderName
+	msgProto.Attachments = attachments
+	eventID := fmt.Sprintf("msg-%s", msg.MessageID())
+
+	var resp iamv1.StreamChatEventsResponse
+	resp.EventId = eventID
+	switch eventType {
+	case "message_received":
+		resp.Payload = &iamv1.StreamChatEventsResponse_MessageReceived{
+			MessageReceived: &iamv1.MessageEvent{ConversationId: conv.ID().String(), Message: msgProto},
+		}
+	case "message_edited":
+		resp.Payload = &iamv1.StreamChatEventsResponse_MessageEdited{
+			MessageEdited: &iamv1.MessageEvent{ConversationId: conv.ID().String(), Message: msgProto},
+		}
+	}
+
+	for _, part := range conv.Participants() {
+		if !part.IsActive() {
+			continue
+		}
+		broadcaster.Publish(&chatinfra.Event{
+			EventID:  fmt.Sprintf("%s-%s", eventID, part.UserID()),
+			UserID:   part.UserID(),
+			Response: &resp,
+		})
+	}
+}
+
+// attachmentToProto maps a domain attachment to its proto representation.
+func attachmentToProto(a *domainChat.Attachment) *iamv1.AttachmentProto {
+	return &iamv1.AttachmentProto{
+		AttachmentId: a.AttachmentID().String(),
+		FileName:     a.FileName(),
+		FileUrl:      a.FileURL(),
+		ContentType:  a.ContentType(),
+		FileSize:     a.FileSize(),
+		ThumbnailUrl: a.ThumbnailURL(),
+	}
+}
+
+func domainMsgToProto(msg *domainChat.Message, plainBody string) *iamv1.MessageProto {
+	replyTo := ""
+	if msg.ReplyToID() != uuid.Nil {
+		replyTo = msg.ReplyToID().String()
+	}
+	return &iamv1.MessageProto{
+		MessageId:      msg.MessageID().String(),
+		ConversationId: msg.ConversationID().String(),
+		SenderUserId:   msg.SenderUserID().String(),
+		Body:           plainBody,
+		IsEdited:       msg.IsEdited(),
+		IsDeleted:      msg.IsDeleted(),
+		ReplyToId:      replyTo,
+		CreatedAt:      msg.CreatedAt().UTC().Format(time.RFC3339Nano),
+		UpdatedAt:      msg.UpdatedAt().UTC().Format(time.RFC3339Nano),
+	}
+}

services/iam/internal/application/chat/clear_history_handler.go

@@ -0,0 +1,34 @@
+package chat
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/google/uuid"
+
+	"github.com/mutugading/goapps-backend/services/iam/internal/domain/chat"
+)
+
+// ClearHistoryHandler clears the calling user's own view of a conversation's
+// message history. Messages remain visible to other participants.
+type ClearHistoryHandler struct {
+	convRepo chat.ConversationRepository
+}
+
+// NewClearHistoryHandler constructs the handler.
+func NewClearHistoryHandler(convRepo chat.ConversationRepository) *ClearHistoryHandler {
+	return &ClearHistoryHandler{convRepo: convRepo}
+}
+
+// Handle verifies callerID is an active participant, then clears their history.
+func (h *ClearHistoryHandler) Handle(ctx context.Context, callerID, convID uuid.UUID) error {
+	conv, err := h.convRepo.GetByID(ctx, convID)
+	if err != nil {
+		return err
+	}
+	p := conv.FindParticipant(callerID)
+	if p == nil || !p.IsActive() {
+		return fmt.Errorf("clear history: %w", chat.ErrNotParticipant)
+	}
+	return h.convRepo.ClearHistory(ctx, convID, callerID)
+}