কেস স্টাডি: স্কেলে Chat System

package main

import (
	"fmt"
	"math"
	"strings"
	"sync"
	"time"
)

// --- Lamport Clock ---
type LamportClock struct {
	mu      sync.Mutex
	counter int64
}

func (lc *LamportClock) Tick() int64 {
	lc.mu.Lock()
	defer lc.mu.Unlock()
	lc.counter++
	return lc.counter
}

func (lc *LamportClock) Update(received int64) int64 {
	lc.mu.Lock()
	defer lc.mu.Unlock()
	if received > lc.counter {
		lc.counter = received
	}
	lc.counter++
	return lc.counter
}

// --- Types ---
type MessageStatus string

const (
	StatusSent      MessageStatus = "sent"
	StatusDelivered MessageStatus = "delivered"
	StatusRead      MessageStatus = "read"
)

type Message struct {
	ID             string        `json:"id"`
	ConversationID string        `json:"conversationId"`
	SenderID       string        `json:"senderId"`
	Content        string        `json:"content"`
	Timestamp      int64         `json:"timestamp"`
	LamportTs      int64         `json:"lamportTs"`
	Status         MessageStatus `json:"status"`
}

type Conversation struct {
	ID            string   `json:"id"`
	Participants  []string `json:"participants"`
	LastMessageAt int64    `json:"lastMessageAt"`
	CreatedAt     int64    `json:"createdAt"`
}

type ReadReceipt struct {
	UserID            string `json:"userId"`
	ConversationID    string `json:"conversationId"`
	LastReadMessageID string `json:"lastReadMessageId"`
	LastReadTimestamp  int64  `json:"lastReadTimestamp"`
}

type TypingEvent struct {
	UserID         string `json:"userId"`
	ConversationID string `json:"conversationId"`
	IsTyping       bool   `json:"isTyping"`
	Timestamp      int64  `json:"timestamp"`
}

type ChatEvent struct {
	Type string
	Data interface{}
}

// --- User Connection ---
type UserConnection struct {
	UserID   string
	Online   bool
	LastSeen int64
	SendFn   func(ChatEvent)
}

// --- Message Store ---
type MessageStore struct {
	mu            sync.RWMutex
	messages      map[string][]Message      // conversationId -> messages
	conversations map[string]*Conversation
	readReceipts  map[string]*ReadReceipt   // "userId:convId" -> receipt
	nextConvID    int
}

func NewMessageStore() *MessageStore {
	return &MessageStore{
		messages:      make(map[string][]Message),
		conversations: make(map[string]*Conversation),
		readReceipts:  make(map[string]*ReadReceipt),
	}
}

func (ms *MessageStore) CreateConversation(participants []string) *Conversation {
	ms.mu.Lock()
	defer ms.mu.Unlock()

	ms.nextConvID++
	id := fmt.Sprintf("conv-%d", ms.nextConvID)
	conv := &Conversation{
		ID:            id,
		Participants:  append([]string{}, participants...),
		LastMessageAt: 0,
		CreatedAt:     time.Now().UnixMilli(),
	}
	ms.conversations[id] = conv
	ms.messages[id] = []Message{}
	fmt.Printf("[STORE] Created conversation %s with participants: %s\n", id, strings.Join(participants, ", "))
	return conv
}

func (ms *MessageStore) StoreMessage(msg Message) {
	ms.mu.Lock()
	defer ms.mu.Unlock()

	ms.messages[msg.ConversationID] = append(ms.messages[msg.ConversationID], msg)
	if conv, ok := ms.conversations[msg.ConversationID]; ok {
		conv.LastMessageAt = msg.Timestamp
	}
	fmt.Printf("[STORE] Stored message %s in %s (lamport: %d)\n", msg.ID, msg.ConversationID, msg.LamportTs)
}

func (ms *MessageStore) GetMessages(conversationID string, limit int) []Message {
	ms.mu.RLock()
	defer ms.mu.RUnlock()

	msgs := ms.messages[conversationID]
	if len(msgs) == 0 {
		return nil
	}

	start := 0
	if len(msgs) > limit {
		start = len(msgs) - limit
	}

	result := make([]Message, len(msgs[start:]))
	copy(result, msgs[start:])

	// Sort by Lamport timestamp
	for i := 0; i < len(result)-1; i++ {
		for j := i + 1; j < len(result); j++ {
			if result[i].LamportTs > result[j].LamportTs {
				result[i], result[j] = result[j], result[i]
			}
		}
	}
	return result
}

func (ms *MessageStore) GetConversation(id string) *Conversation {
	ms.mu.RLock()
	defer ms.mu.RUnlock()
	return ms.conversations[id]
}

func (ms *MessageStore) GetConversationsForUser(userID string) []*Conversation {
	ms.mu.RLock()
	defer ms.mu.RUnlock()

	var result []*Conversation
	for _, conv := range ms.conversations {
		for _, p := range conv.Participants {
			if p == userID {
				result = append(result, conv)
				break
			}
		}
	}
	return result
}

func (ms *MessageStore) SetReadReceipt(userID, conversationID, messageID string) *ReadReceipt {
	ms.mu.Lock()
	defer ms.mu.Unlock()

	key := fmt.Sprintf("%s:%s", userID, conversationID)
	receipt := &ReadReceipt{
		UserID:            userID,
		ConversationID:    conversationID,
		LastReadMessageID: messageID,
		LastReadTimestamp:  time.Now().UnixMilli(),
	}
	ms.readReceipts[key] = receipt

	// Mark messages as read
	msgs := ms.messages[conversationID]
	for i := range msgs {
		if msgs[i].SenderID != userID && msgs[i].Status != StatusRead {
			msgs[i].Status = StatusRead
		}
		if msgs[i].ID == messageID {
			msgs[i].Status = StatusRead
			break
		}
	}
	return receipt
}

func (ms *MessageStore) MarkDelivered(conversationID, messageID string) {
	ms.mu.Lock()
	defer ms.mu.Unlock()

	msgs := ms.messages[conversationID]
	for i := range msgs {
		if msgs[i].ID == messageID && msgs[i].Status == StatusSent {
			msgs[i].Status = StatusDelivered
			return
		}
	}
}

// --- Offline Queue ---
type OfflineQueue struct {
	mu     sync.Mutex
	queues map[string][]Message // userId -> pending messages
}

func NewOfflineQueue() *OfflineQueue {
	return &OfflineQueue{queues: make(map[string][]Message)}
}

func (oq *OfflineQueue) Enqueue(userID string, msg Message) {
	oq.mu.Lock()
	defer oq.mu.Unlock()
	oq.queues[userID] = append(oq.queues[userID], msg)
	fmt.Printf("[OFFLINE] Queued message %s for offline user %s\n", msg.ID, userID)
}

func (oq *OfflineQueue) Drain(userID string) []Message {
	oq.mu.Lock()
	defer oq.mu.Unlock()
	msgs := oq.queues[userID]
	delete(oq.queues, userID)
	if len(msgs) > 0 {
		fmt.Printf("[OFFLINE] Draining %d messages for user %s\n", len(msgs), userID)
	}
	return msgs
}

// --- Connection Manager ---
type ConnectionMgr struct {
	mu    sync.RWMutex
	conns map[string]*UserConnection
}

func NewConnectionMgr() *ConnectionMgr {
	return &ConnectionMgr{conns: make(map[string]*UserConnection)}
}

func (cm *ConnectionMgr) Connect(userID string, sendFn func(ChatEvent)) *UserConnection {
	cm.mu.Lock()
	defer cm.mu.Unlock()
	conn := &UserConnection{
		UserID:   userID,
		Online:   true,
		LastSeen: time.Now().UnixMilli(),
		SendFn:   sendFn,
	}
	cm.conns[userID] = conn
	fmt.Printf("[PRESENCE] User %s is now ONLINE\n", userID)
	return conn
}

func (cm *ConnectionMgr) Disconnect(userID string) {
	cm.mu.Lock()
	defer cm.mu.Unlock()
	if conn, ok := cm.conns[userID]; ok {
		conn.Online = false
		conn.LastSeen = time.Now().UnixMilli()
		fmt.Printf("[PRESENCE] User %s is now OFFLINE\n", userID)
	}
}

func (cm *ConnectionMgr) GetConnection(userID string) *UserConnection {
	cm.mu.RLock()
	defer cm.mu.RUnlock()
	conn := cm.conns[userID]
	if conn != nil && conn.Online {
		return conn
	}
	return nil
}

func (cm *ConnectionMgr) IsOnline(userID string) bool {
	cm.mu.RLock()
	defer cm.mu.RUnlock()
	conn := cm.conns[userID]
	return conn != nil && conn.Online
}

func (cm *ConnectionMgr) GetOnlineUsers() []string {
	cm.mu.RLock()
	defer cm.mu.RUnlock()
	var online []string
	for uid, conn := range cm.conns {
		if conn.Online {
			online = append(online, uid)
		}
	}
	return online
}

// --- Fan-out Service ---
type FanOutService struct {
	store        *MessageStore
	connMgr      *ConnectionMgr
	offlineQueue *OfflineQueue
	clock        *LamportClock
	nextMsgID    int
	mu           sync.Mutex
}

func NewFanOutService(store *MessageStore, connMgr *ConnectionMgr, oq *OfflineQueue, clock *LamportClock) *FanOutService {
	return &FanOutService{
		store:        store,
		connMgr:      connMgr,
		offlineQueue: oq,
		clock:        clock,
	}
}

func (fs *FanOutService) genMsgID() string {
	fs.mu.Lock()
	defer fs.mu.Unlock()
	fs.nextMsgID++
	return fmt.Sprintf("msg-%d", fs.nextMsgID)
}

func (fs *FanOutService) SendMessage(senderID, conversationID, content string) *Message {
	conv := fs.store.GetConversation(conversationID)
	if conv == nil {
		fmt.Printf("[ERROR] Conversation %s not found\n", conversationID)
		return nil
	}

	// Truncate content
	if len(content) > 4096 {
		content = content[:4096]
	}

	msg := Message{
		ID:             fs.genMsgID(),
		ConversationID: conversationID,
		SenderID:       senderID,
		Content:        content,
		Timestamp:      time.Now().UnixMilli(),
		LamportTs:      fs.clock.Tick(),
		Status:         StatusSent,
	}

	fs.store.StoreMessage(msg)

	fmt.Printf("[FANOUT] Distributing message %s to %d participants\n", msg.ID, len(conv.Participants))

	for _, participantID := range conv.Participants {
		if participantID == senderID {
			continue
		}

		conn := fs.connMgr.GetConnection(participantID)
		if conn != nil {
			conn.SendFn(ChatEvent{Type: "message", Data: msg})
			fs.store.MarkDelivered(conversationID, msg.ID)
			fmt.Printf("[FANOUT] Delivered %s to %s (online)\n", msg.ID, participantID)

			if senderConn := fs.connMgr.GetConnection(senderID); senderConn != nil {
				senderConn.SendFn(ChatEvent{
					Type: "delivered",
					Data: map[string]string{"messageId": msg.ID, "conversationId": conversationID},
				})
			}
		} else {
			fs.offlineQueue.Enqueue(participantID, msg)
		}
	}

	return &msg
}

func (fs *FanOutService) SendReadReceipt(userID, conversationID, messageID string) {
	receipt := fs.store.SetReadReceipt(userID, conversationID, messageID)
	conv := fs.store.GetConversation(conversationID)
	if conv == nil {
		return
	}

	for _, participantID := range conv.Participants {
		if participantID == userID {
			continue
		}
		if conn := fs.connMgr.GetConnection(participantID); conn != nil {
			conn.SendFn(ChatEvent{Type: "read_receipt", Data: receipt})
			fmt.Printf("[RECEIPT] Sent read receipt to %s (read by %s)\n", participantID, userID)
		}
	}
}

func (fs *FanOutService) SendTypingIndicator(userID, conversationID string, isTyping bool) {
	conv := fs.store.GetConversation(conversationID)
	if conv == nil {
		return
	}

	event := TypingEvent{
		UserID:         userID,
		ConversationID: conversationID,
		IsTyping:       isTyping,
		Timestamp:      time.Now().UnixMilli(),
	}

	for _, participantID := range conv.Participants {
		if participantID == userID {
			continue
		}
		if conn := fs.connMgr.GetConnection(participantID); conn != nil {
			conn.SendFn(ChatEvent{Type: "typing", Data: event})
		}
	}
}

func (fs *FanOutService) HandleReconnect(userID string) {
	pending := fs.offlineQueue.Drain(userID)
	if len(pending) == 0 {
		return
	}

	conn := fs.connMgr.GetConnection(userID)
	if conn == nil {
		return
	}

	conn.SendFn(ChatEvent{Type: "offline_messages", Data: pending})

	for _, msg := range pending {
		fs.store.MarkDelivered(msg.ConversationID, msg.ID)
		if senderConn := fs.connMgr.GetConnection(msg.SenderID); senderConn != nil {
			senderConn.SendFn(ChatEvent{
				Type: "delivered",
				Data: map[string]string{"messageId": msg.ID, "conversationId": msg.ConversationID},
			})
		}
	}
	fmt.Printf("[RECONNECT] Delivered %d offline messages to %s\n", len(pending), userID)
}

func (fs *FanOutService) BroadcastPresence(userID string, online bool) {
	conversations := fs.store.GetConversationsForUser(userID)
	notified := make(map[string]bool)

	for _, conv := range conversations {
		for _, participantID := range conv.Participants {
			if participantID == userID || notified[participantID] {
				continue
			}
			notified[participantID] = true

			if conn := fs.connMgr.GetConnection(participantID); conn != nil {
				conn.SendFn(ChatEvent{
					Type: "presence",
					Data: map[string]interface{}{"userId": userID, "online": online},
				})
			}
		}
	}
}

// Suppress unused import warning
var _ = math.MaxFloat64

// --- Main ---
func main() {
	store := NewMessageStore()
	connMgr := NewConnectionMgr()
	offlineQueue := NewOfflineQueue()
	clock := &LamportClock{}
	fanout := NewFanOutService(store, connMgr, offlineQueue, clock)

	type eventEntry struct {
		User  string
		Event ChatEvent
	}
	var eventLog []eventEntry
	var logMu sync.Mutex

	makeSendFn := func(userID string) func(ChatEvent) {
		return func(e ChatEvent) {
			logMu.Lock()
			eventLog = append(eventLog, eventEntry{User: userID, Event: e})
			logMu.Unlock()
			fmt.Printf("  >> [%s] received %s\n", userID, e.Type)
		}
	}

	fmt.Println("=== Setup ===")
	connMgr.Connect("fatima", makeSendFn("fatima"))
	connMgr.Connect("omar", makeSendFn("omar"))
	// yusuf starts offline

	conv1 := store.CreateConversation([]string{"fatima", "omar", "yusuf"})

	fmt.Println("\n=== Fatima sends a message ===")
	msg1 := fanout.SendMessage("fatima", conv1.ID, "Hey team, how is everyone?")

	fmt.Println("\n=== Omar reads the message ===")
	fanout.SendReadReceipt("omar", conv1.ID, msg1.ID)

	fmt.Println("\n=== Omar types a reply ===")
	fanout.SendTypingIndicator("omar", conv1.ID, true)
	msg2 := fanout.SendMessage("omar", conv1.ID, "Doing great! Working on the new feature.")
	fanout.SendTypingIndicator("omar", conv1.ID, false)

	fmt.Println("\n=== Yusuf comes online ===")
	connMgr.Connect("yusuf", makeSendFn("yusuf"))
	fanout.HandleReconnect("yusuf")
	fanout.BroadcastPresence("yusuf", true)

	fmt.Println("\n=== Yusuf reads messages ===")
	fanout.SendReadReceipt("yusuf", conv1.ID, msg2.ID)

	fmt.Println("\n=== Final message history ===")
	messages := store.GetMessages(conv1.ID, 50)
	for _, msg := range messages {
		fmt.Printf("  [%d] %s: %s (%s)\n", msg.LamportTs, msg.SenderID, msg.Content, msg.Status)
	}

	fmt.Println("\n=== Online users ===")
	fmt.Printf("  %s\n", strings.Join(connMgr.GetOnlineUsers(), ", "))

	fmt.Printf("\n=== Event log (%d events) ===\n", len(eventLog))
	for _, entry := range eventLog {
		fmt.Printf("  %s: %s\n", entry.User, entry.Event.Type)
	}
}