message-center/connector_whatsapp.go

package main

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"
)

// WhatsAppConnector wraps the message-bridge HTTP API
type WhatsAppConnector struct {
	name     string
	baseURL  string
	ctx      chan struct{}
	callback func()
	mu       sync.Mutex
	lastSeen map[string]bool // Track seen message IDs
}

// WhatsAppMessage is the format from message-bridge
type WhatsAppMessage struct {
	ID            string    `json:"id"`
	Platform      string    `json:"platform"`
	From          string    `json:"from"`
	FromName      string    `json:"from_name,omitempty"`
	To            string    `json:"to,omitempty"`
	Body          string    `json:"body"`
	Timestamp     time.Time `json:"timestamp"`
	IsGroup       bool      `json:"is_group"`
	GroupName     string    `json:"group_name,omitempty"`
	MediaType     string    `json:"media_type,omitempty"`
	MediaURL      string    `json:"media_url,omitempty"`
	MediaPath     string    `json:"media_path,omitempty"`
	HasMedia      bool      `json:"has_media"`
	Transcription string    `json:"transcription,omitempty"`
}

// NewWhatsAppConnector creates a new WhatsApp connector
func NewWhatsAppConnector(name, baseURL string) *WhatsAppConnector {
	return &WhatsAppConnector{
		name:     name,
		baseURL:  baseURL,
		lastSeen: make(map[string]bool),
	}
}

func (w *WhatsAppConnector) Name() string {
	return w.name
}

func (w *WhatsAppConnector) fetchMessages() ([]WhatsAppMessage, error) {
	resp, err := http.Get(w.baseURL + "/messages")
	if err != nil {
		return nil, fmt.Errorf("fetch messages: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode != 200 {
		body, _ := io.ReadAll(resp.Body)
		return nil, fmt.Errorf("fetch messages: status %d: %s", resp.StatusCode, string(body))
	}

	var messages []WhatsAppMessage
	if err := json.NewDecoder(resp.Body).Decode(&messages); err != nil {
		return nil, fmt.Errorf("decode messages: %w", err)
	}

	return messages, nil
}

func (w *WhatsAppConnector) convertMessage(msg WhatsAppMessage) UnifiedMessage {
	um := UnifiedMessage{
		ID:          fmt.Sprintf("%s:%s", w.name, msg.ID),
		Source:      w.name,
		SourceUID:   msg.ID,
		From:        msg.From,
		FromName:    msg.FromName,
		Timestamp:   msg.Timestamp,
		Body:        msg.Body,
		Seen:        false, // WhatsApp doesn't have a seen concept in our bridge
		Attachments: []AttachmentMeta{},
	}

	// Handle group vs direct
	if msg.IsGroup {
		um.To = msg.GroupName
		if um.To == "" {
			um.To = msg.To
		}
	} else {
		// For direct messages, "to" is the recipient (us)
		um.To = msg.To
	}

	// Use transcription as body for voice messages if available
	if msg.Transcription != "" && msg.Body == "" {
		um.Body = msg.Transcription
	}

	// Handle media as attachment
	if msg.HasMedia && msg.MediaPath != "" {
		filename := filepath.Base(msg.MediaPath)
		mimeType := "application/octet-stream"
		switch msg.MediaType {
		case "image":
			mimeType = "image/jpeg"
		case "video":
			mimeType = "video/mp4"
		case "voice", "audio":
			mimeType = "audio/ogg"
		case "document":
			mimeType = "application/pdf"
		}

		// Get file size
		var size int
		if fi, err := os.Stat(msg.MediaPath); err == nil {
			size = int(fi.Size())
		}

		um.Attachments = append(um.Attachments, AttachmentMeta{
			Name: filename,
			Mime: mimeType,
			Size: size,
		})

		// Store media path for later retrieval
		um.SourceExtra = msg.MediaPath
	}

	return um
}

func (w *WhatsAppConnector) FetchNew() ([]UnifiedMessage, error) {
	messages, err := w.fetchMessages()
	if err != nil {
		return nil, err
	}

	w.mu.Lock()
	defer w.mu.Unlock()

	var result []UnifiedMessage
	for _, msg := range messages {
		// Skip if already marked as seen
		if w.lastSeen[msg.ID] {
			continue
		}
		// Filter to last 24 hours for "new"
		if time.Since(msg.Timestamp) < 24*time.Hour {
			result = append(result, w.convertMessage(msg))
		}
	}

	return result, nil
}

func (w *WhatsAppConnector) FetchSince(since time.Time) ([]UnifiedMessage, error) {
	messages, err := w.fetchMessages()
	if err != nil {
		return nil, err
	}

	var result []UnifiedMessage
	for _, msg := range messages {
		if msg.Timestamp.After(since) || msg.Timestamp.Equal(since) {
			result = append(result, w.convertMessage(msg))
		}
	}

	return result, nil
}

func (w *WhatsAppConnector) FetchOne(sourceID string) (*UnifiedMessage, error) {
	messages, err := w.fetchMessages()
	if err != nil {
		return nil, err
	}

	for _, msg := range messages {
		if msg.ID == sourceID {
			um := w.convertMessage(msg)
			return &um, nil
		}
	}

	return nil, fmt.Errorf("message not found: %s", sourceID)
}

func (w *WhatsAppConnector) Archive(sourceID string) error {
	// WhatsApp doesn't support archiving individual messages
	// Mark it as seen in our local tracking
	w.mu.Lock()
	w.lastSeen[sourceID] = true
	w.mu.Unlock()
	return nil
}

func (w *WhatsAppConnector) Delete(sourceID string) error {
	// WhatsApp doesn't support deleting messages via API
	return fmt.Errorf("WhatsApp does not support message deletion")
}

func (w *WhatsAppConnector) Reply(sourceID string, body string, attachments []string) error {
	// Get original message to find the chat
	msg, err := w.FetchOne(sourceID)
	if err != nil {
		return err
	}

	// Build send request
	sendReq := map[string]interface{}{
		"to":   msg.From, // Reply to sender
		"body": body,
	}

	// Handle attachments if any
	if len(attachments) > 0 {
		sendReq["media"] = attachments[0] // message-bridge only supports one attachment
	}

	reqBody, _ := json.Marshal(sendReq)
	resp, err := http.Post(w.baseURL+"/send", "application/json", bytes.NewReader(reqBody))
	if err != nil {
		return fmt.Errorf("send reply: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode >= 400 {
		body, _ := io.ReadAll(resp.Body)
		return fmt.Errorf("send reply: status %d: %s", resp.StatusCode, string(body))
	}

	return nil
}

func (w *WhatsAppConnector) MarkSeen(sourceID string) error {
	// Track locally
	w.mu.Lock()
	w.lastSeen[sourceID] = true
	w.mu.Unlock()
	return nil
}

func (w *WhatsAppConnector) GetAttachment(sourceID string, filename string) ([]byte, error) {
	// First, find the message to get the media path
	messages, err := w.fetchMessages()
	if err != nil {
		return nil, err
	}

	for _, msg := range messages {
		if msg.ID == sourceID && msg.MediaPath != "" {
			// Check if this matches the requested filename
			if strings.HasSuffix(msg.MediaPath, filename) || filepath.Base(msg.MediaPath) == filename {
				return os.ReadFile(msg.MediaPath)
			}
		}
	}

	// Try fetching from message-bridge media endpoint
	resp, err := http.Get(w.baseURL + "/media/" + filename)
	if err != nil {
		return nil, fmt.Errorf("fetch media: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode != 200 {
		return nil, fmt.Errorf("media not found: %s", filename)
	}

	return io.ReadAll(resp.Body)
}

func (w *WhatsAppConnector) Start(callback func()) error {
	w.mu.Lock()
	defer w.mu.Unlock()

	w.ctx = make(chan struct{})
	w.callback = callback

	// Poll for new messages periodically
	go w.pollLoop()

	return nil
}

func (w *WhatsAppConnector) pollLoop() {
	ticker := time.NewTicker(10 * time.Second)
	defer ticker.Stop()

	// Initial fetch to populate seen set
	msgs, err := w.fetchMessages()
	if err == nil {
		w.mu.Lock()
		for _, msg := range msgs {
			w.lastSeen[msg.ID] = true
		}
		w.mu.Unlock()
	}

	log.Printf("[%s] Started polling for new messages", w.name)

	for {
		select {
		case <-w.ctx:
			return
		case <-ticker.C:
			w.checkForNew()
		}
	}
}

func (w *WhatsAppConnector) checkForNew() {
	msgs, err := w.fetchMessages()
	if err != nil {
		log.Printf("[%s] Poll error: %v", w.name, err)
		return
	}

	hasNew := false
	w.mu.Lock()
	for _, msg := range msgs {
		if !w.lastSeen[msg.ID] {
			hasNew = true
			w.lastSeen[msg.ID] = true
			log.Printf("[%s] New message from %s: %s", w.name, msg.FromName, truncateStr(msg.Body, 50))
		}
	}
	callback := w.callback
	w.mu.Unlock()

	if hasNew && callback != nil {
		go callback()
	}
}

func truncateStr(s string, maxLen int) string {
	if len(s) <= maxLen {
		return s
	}
	return s[:maxLen] + "..."
}

func (w *WhatsAppConnector) Stop() {
	w.mu.Lock()
	defer w.mu.Unlock()

	if w.ctx != nil {
		close(w.ctx)
	}
}