workspace/apps/go-crm/internal/whatsapp/whatsmeow.go

package whatsapp

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"

	_ "github.com/glebarez/go-sqlite"
	"go.mau.fi/whatsmeow"
	"go.mau.fi/whatsmeow/store/sqlstore"
	"go.mau.fi/whatsmeow/types"
	"go.mau.fi/whatsmeow/types/events"
	waLog "go.mau.fi/whatsmeow/util/log"
)

type WhatsmeowAdapter struct {
	mu             sync.RWMutex
	clients        map[int64]*whatsmeow.Client
	qrChans        map[int64]chan QRFrame // persistent QR channel per client, reused across polls
	container      *sqlstore.Container
	db             *sql.DB
	goEndpoint     string
	internalSecret string
	appCtx         context.Context
}

func NewWhatsmeowAdapter(storePath, goEndpoint, internalSecret string) (*WhatsmeowAdapter, error) {
	db, err := sql.Open("sqlite", storePath)
	if err != nil {
		return nil, fmt.Errorf("failed to open database: %w", err)
	}
	_, err = db.Exec("PRAGMA foreign_keys = ON")
	if err != nil {
		return nil, fmt.Errorf("failed to enable foreign keys: %w", err)
	}
	container := sqlstore.NewWithDB(db, "sqlite", waLog.Stdout("SQL", "DEBUG", true))
	if err = container.Upgrade(context.Background()); err != nil {
		return nil, fmt.Errorf("failed to upgrade database: %w", err)
	}

	return &WhatsmeowAdapter{
		clients:        make(map[int64]*whatsmeow.Client),
		qrChans:        make(map[int64]chan QRFrame),
		container:      container,
		db:             db,
		goEndpoint:     goEndpoint,
		internalSecret: internalSecret,
		appCtx:         context.Background(),
	}, nil
}

func (a *WhatsmeowAdapter) Connect(ctx context.Context, clientID int64) (<-chan QRFrame, error) {
	a.mu.Lock()

	// Already authenticated — return immediately.
	if client, ok := a.clients[clientID]; ok && client.IsLoggedIn() {
		a.mu.Unlock()
		ch := make(chan QRFrame, 1)
		ch <- QRFrame{State: StateConnected}
		close(ch)
		return ch, nil
	}

	// QR session already in progress — return the same persistent channel.
	// The goroutine keeps writing QR codes to it; each poll reads the latest one.
	if ch, ok := a.qrChans[clientID]; ok {
		a.mu.Unlock()
		log.Printf("WA-Connect: Reusing existing QR session for client %d", clientID)
		return ch, nil
	}

	// Read stored JID before releasing lock.
	jid, _ := a.getJIDFromDBLocked(clientID)
	a.mu.Unlock()

	// Try to resume a previously paired session (no QR needed).
	if jid != "" {
		if ch, err := a.resumeSession(clientID, jid); err == nil {
			return ch, nil
		}
		log.Printf("WA-Connect: Resume failed for client %d, starting QR flow", clientID)
		// Clear stale JID so we don't retry resume on every poll.
		a.clearJIDFromDB(clientID)
	}

	// Start a fresh QR session — all network I/O, no lock held.
	return a.startQRSession(clientID)
}

// resumeSession attempts to reconnect a previously paired device using its stored JID.
func (a *WhatsmeowAdapter) resumeSession(clientID int64, jid string) (<-chan QRFrame, error) {
	parsedJID, err := types.ParseJID(jid)
	if err != nil {
		return nil, err
	}
	device, err := a.container.GetDevice(a.appCtx, parsedJID)
	if err != nil || device == nil {
		return nil, fmt.Errorf("device not found")
	}

	client := whatsmeow.NewClient(device, waLog.Stdout("Client-"+strconv.FormatInt(clientID, 10), "DEBUG", true))

	a.mu.Lock()
	a.clients[clientID] = client
	a.mu.Unlock()

	if err := client.Connect(); err != nil {
		a.mu.Lock()
		delete(a.clients, clientID)
		a.mu.Unlock()
		return nil, err
	}

	a.addMessageHandler(client, clientID)
	ch := make(chan QRFrame, 1)
	ch <- QRFrame{State: StateConnected}
	close(ch)
	return ch, nil
}

// startQRSession creates a fresh device, initiates the WhatsApp WebSocket connection,
// and returns a persistent channel that emits QR frames. All network I/O without holding mutex.
// The channel is stored in a.qrChans so subsequent polls reuse it without restarting the session.
func (a *WhatsmeowAdapter) startQRSession(clientID int64) (<-chan QRFrame, error) {
	device := a.container.NewDevice()
	log.Printf("WA-Connect: Created device for client %d", clientID)
	client := whatsmeow.NewClient(device, waLog.Stdout("Client-"+strconv.FormatInt(clientID, 10), "DEBUG", true))

	// GetQRChannel must be called before Connect().
	log.Printf("WA-Connect: Getting QR channel for client %d", clientID)
	qrChan, err := client.GetQRChannel(a.appCtx)
	if err != nil {
		log.Printf("WA-Connect: GetQRChannel error for client %d: %v", clientID, err)
		return nil, fmt.Errorf("failed to get QR channel: %w", err)
	}

	// ch is buffered so QR codes accumulate; polls read the latest available frame.
	// Size 20 = enough to hold multiple rotated QR codes without blocking the goroutine.
	ch := make(chan QRFrame, 20)

	a.mu.Lock()
	a.clients[clientID] = client
	a.qrChans[clientID] = ch
	a.mu.Unlock()

	connectErr := make(chan error, 1)

	log.Printf("WA-Connect: Starting Connect() goroutine for client %d", clientID)
	go func() {
		if err := client.Connect(); err != nil {
			log.Printf("WA-Connect: Connect() error for client %d: %v", clientID, err)
			connectErr <- err
			return
		}
		log.Printf("WA-Connect: Connect() succeeded for client %d", clientID)
		connectErr <- nil
	}()

	// Wait for WS handshake — lock NOT held, no deadlock risk.
	select {
	case err := <-connectErr:
		if err != nil {
			a.mu.Lock()
			delete(a.clients, clientID)
			delete(a.qrChans, clientID)
			a.mu.Unlock()
			return nil, fmt.Errorf("failed to connect: %w", err)
		}
	case <-time.After(15 * time.Second):
		log.Printf("WA-Connect: Connect() timeout for client %d", clientID)
		a.mu.Lock()
		delete(a.clients, clientID)
		delete(a.qrChans, clientID)
		a.mu.Unlock()
		return nil, fmt.Errorf("connection timeout")
	}

	// Fan-out goroutine: translates whatsmeow events into QRFrames on the persistent ch.
	// Outlives any HTTP request. Cleans up qrChans entry on terminal events.
	go func() {
		defer func() {
			a.mu.Lock()
			delete(a.qrChans, clientID)
			a.mu.Unlock()
			close(ch)
		}()
		for {
			select {
			case evt, ok := <-qrChan:
				if !ok {
					return
				}
				switch evt.Event {
				case whatsmeow.QRChannelEventCode:
					// Drain stale QR frames so the buffer holds only the latest code.
					for len(ch) > 0 {
						<-ch
					}
					ch <- QRFrame{QR: evt.Code, State: StateWaitingQR}
					log.Printf("WA-Connect: QR code updated for client %d", clientID)
				case whatsmeow.QRChannelSuccess.Event:
					go a.syncContacts(client, clientID)
					go a.saveJIDOnConnect(client, clientID)
					ch <- QRFrame{State: StateConnected}
					return
				case whatsmeow.QRChannelEventError:
					errMsg := "pairing error"
					if evt.Error != nil {
						errMsg = evt.Error.Error()
					}
					ch <- QRFrame{State: StateFailed, Error: errMsg}
					return
				case whatsmeow.QRChannelTimeout.Event:
					ch <- QRFrame{State: StateFailed, Error: "QR code timed out"}
					return
				case whatsmeow.QRChannelClientOutdated.Event:
					ch <- QRFrame{State: StateFailed, Error: "client outdated"}
					return
				default:
					ch <- QRFrame{State: StateFailed, Error: "unexpected event: " + evt.Event}
					return
				}
			case <-a.appCtx.Done():
				return
			}
		}
	}()

	a.addMessageHandler(client, clientID)

	return ch, nil
}

func (a *WhatsmeowAdapter) addMessageHandler(client *whatsmeow.Client, clientID int64) {
	log.Printf("[Client-%d] Registering message handler", clientID)
	client.AddEventHandler(func(evt interface{}) {
		log.Printf("[Client-%d] Event received: %T", clientID, evt)
		msg, ok := evt.(*events.Message)
		if !ok {
			return
		}
		log.Printf("[Client-%d] Message event: IsFromMe=%v, Sender=%v, Chat=%v", clientID, msg.Info.IsFromMe, msg.Info.Sender, msg.Info.Chat)

		// Skip messages from our own device.
		// Compare sender JID with device JID (LID messages may have IsFromMe=true incorrectly).
		if client.Store.ID != nil && msg.Info.Sender.User == client.Store.ID.User {
			log.Printf("[Client-%d] Skipping message from own device", clientID)
			return
		}

		// Prefer phone JID (SenderAlt) when sender is LID.
		sender := msg.Info.Sender
		if sender.Server == types.HiddenUserServer && !msg.Info.SenderAlt.IsEmpty() && msg.Info.SenderAlt.Server == types.DefaultUserServer {
			sender = msg.Info.SenderAlt
		}
		phone := sender.String()
		pushName := msg.Info.PushName

		// Extract text body from the message.
		var text string
		if msg.Message != nil {
			if c := msg.Message.GetConversation(); c != "" {
				text = c
			} else if ext := msg.Message.GetExtendedTextMessage(); ext != nil {
				text = ext.GetText()
			}
		}
		log.Printf("[Client-%d] Incoming msg from %s (pushName=%s): text=%q", clientID, phone, pushName, text)

		if phone != "" {
			contact := Contact{
				Phone:   phone,
				Name:    pushName,
				Message: text,
				FromMe:  false,
				Time:    time.Now(),
			}
			a.postContact(a.appCtx, clientID, contact, msg.Info.ID)
		}
	})
}

func (a *WhatsmeowAdapter) syncContacts(client *whatsmeow.Client, clientID int64) {
	// Note: GetChats API changed in newer versions
	// Simplified for build - real implementation would use new API
}

func (a *WhatsmeowAdapter) postContact(ctx context.Context, clientID int64, contact Contact, messageID string) {
	payload := map[string]interface{}{
		"client_id":  clientID,
		"name":       contact.Name,
		"phone":      contact.Phone,
		"message":    contact.Message,
		"message_id": messageID,
	}

	body, _ := json.Marshal(payload)
	httpBody := bytes.NewReader(body)
	req, _ := http.NewRequestWithContext(ctx, "POST", a.goEndpoint+"/leads/ingest", httpBody)
	req.Header.Set("Content-Type", "application/json; charset=utf-8")
	req.Header.Set("X-Internal-Secret", a.internalSecret)

	log.Printf("[Client-%d] postContact payload: %s", clientID, string(body))

	httpClient := &http.Client{Timeout: 10 * time.Second}
	resp, err := httpClient.Do(req)
	if err != nil {
		log.Printf("postContact failed to ingest lead for client %d: %v", clientID, err)
		return
	}
	defer resp.Body.Close()
	if resp.StatusCode >= 400 {
		log.Printf("postContact ingest failed for client %d: status %d", clientID, resp.StatusCode)
	} else {
		log.Printf("[Client-%d] postContact success: status %d", clientID, resp.StatusCode)
	}
}

func (a *WhatsmeowAdapter) getJIDFromDB(clientID int64) (string, error) {
	a.mu.RLock()
	db := a.db
	a.mu.RUnlock()
	return a.queryJID(db, clientID)
}

// getJIDFromDBLocked reads JID from DB; caller must hold a.mu (read or write).
func (a *WhatsmeowAdapter) getJIDFromDBLocked(clientID int64) (string, error) {
	return a.queryJID(a.db, clientID)
}

func (a *WhatsmeowAdapter) queryJID(db *sql.DB, clientID int64) (string, error) {
	if db == nil {
		return "", nil
	}

	var jid string
	err := db.QueryRow("SELECT whatsapp_jid FROM clients WHERE client_id = ?", clientID).Scan(&jid)
	if err == sql.ErrNoRows {
		return "", nil
	}
	if err != nil {
		return "", err
	}
	return jid, nil
}

func (a *WhatsmeowAdapter) getDeviceFromStore(jid string) interface{} {
	return nil
}

func (a *WhatsmeowAdapter) saveJIDOnConnect(client *whatsmeow.Client, clientID int64) {
	if client.Store.ID == nil {
		return
	}
	jid := client.Store.ID.User
	if jid == "" {
		return
	}
	phone := strings.Split(jid, "@")[0]
	_ = a.SaveJID(clientID, jid)
	_ = a.saveWhatsAppNumber(clientID, phone)
	_ = a.markConnected(clientID)
}

func (a *WhatsmeowAdapter) saveWhatsAppNumber(clientID int64, phone string) error {
	a.mu.RLock()
	db := a.db
	a.mu.RUnlock()

	if db == nil {
		return nil
	}

	_, err := db.Exec("UPDATE clients SET whatsapp_number = ? WHERE client_id = ?", phone, clientID)
	return err
}

func (a *WhatsmeowAdapter) markConnected(clientID int64) error {
	a.mu.RLock()
	db := a.db
	a.mu.RUnlock()

	if db == nil {
		return nil
	}

	_, err := db.Exec("UPDATE clients SET whatsapp_connected = 1 WHERE client_id = ?", clientID)
	return err
}

func (a *WhatsmeowAdapter) Disconnect(ctx context.Context, clientID int64) error {
	a.mu.Lock()
	defer a.mu.Unlock()

	client, ok := a.clients[clientID]
	if ok {
		client.Disconnect()
		delete(a.clients, clientID)
	}
	delete(a.qrChans, clientID)
	return nil
}

func (a *WhatsmeowAdapter) clearJIDFromDB(clientID int64) {
	a.mu.RLock()
	db := a.db
	a.mu.RUnlock()
	if db != nil {
		db.Exec("UPDATE clients SET whatsapp_jid = '' WHERE client_id = ?", clientID)
	}
}

func (a *WhatsmeowAdapter) SetClientDB(db interface{}) {
	a.mu.Lock()
	defer a.mu.Unlock()
	a.db = db.(*sql.DB)
}

func (a *WhatsmeowAdapter) SaveJID(clientID int64, jid string) error {
	a.mu.RLock()
	db := a.db
	a.mu.RUnlock()

	if db == nil {
		return nil
	}

	_, err := db.Exec("UPDATE clients SET whatsapp_jid = ? WHERE client_id = ?", jid, clientID)
	return err
}

func (a *WhatsmeowAdapter) GetJID(clientID int64) (string, error) {
	a.mu.RLock()
	db := a.db
	a.mu.RUnlock()

	if db == nil {
		return "", nil
	}

	var jid string
	err := db.QueryRow("SELECT whatsapp_jid FROM clients WHERE client_id = ?", clientID).Scan(&jid)
	if err == sql.ErrNoRows {
		return "", nil
	}
	return jid, err
}

func (a *WhatsmeowAdapter) IsConnected(ctx context.Context, clientID int64) (bool, error) {
	if a == nil {
		return false, fmt.Errorf("adapter not initialized")
	}

	a.mu.RLock()
	client, ok := a.clients[clientID]
	a.mu.RUnlock()

	if ok && client != nil && client.IsLoggedIn() {
		return true, nil
	}

	// No client in memory — try to resume from stored JID.
	jid, err := a.getJIDFromDB(clientID)
	if err != nil || jid == "" {
		return false, nil
	}

	// Attempt to reconnect silently.
	if ch, err := a.Connect(ctx, clientID); err == nil {
		for frame := range ch {
			if frame.State == StateConnected {
				return true, nil
			}
			if frame.State == StateFailed {
				break
			}
		}
	}

	return false, nil
}