420 lines
11 KiB
Go
420 lines
11 KiB
Go
package whatsapp
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log"
|
|
"net/http"
|
|
"strconv"
|
|
"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) {
|
|
client.AddEventHandler(func(evt interface{}) {
|
|
msg, ok := evt.(*events.Message)
|
|
if !ok {
|
|
return
|
|
}
|
|
if msg.Info.IsFromMe {
|
|
return
|
|
}
|
|
phone := msg.Info.Sender.String()
|
|
pushName := msg.Info.PushName
|
|
if phone != "" {
|
|
contact := Contact{
|
|
Phone: phone,
|
|
Name: pushName,
|
|
FromMe: false,
|
|
Time: time.Now(),
|
|
}
|
|
a.postContact(a.appCtx, clientID, contact)
|
|
}
|
|
})
|
|
}
|
|
|
|
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) {
|
|
payload := map[string]interface{}{
|
|
"client_id": clientID,
|
|
"name": contact.Name,
|
|
"phone": contact.Phone,
|
|
}
|
|
|
|
body, _ := json.Marshal(payload)
|
|
httpBody := bytes.NewReader(body)
|
|
req, _ := http.NewRequestWithContext(ctx, "POST", a.goEndpoint+"/customers", httpBody)
|
|
req.Header.Set("Content-Type", "application/json")
|
|
req.Header.Set("X-Internal-Secret", a.internalSecret)
|
|
|
|
httpClient := &http.Client{Timeout: 10 * time.Second}
|
|
httpClient.Do(req)
|
|
}
|
|
|
|
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
|
|
}
|
|
_ = a.SaveJID(clientID, jid)
|
|
_ = a.markConnected(clientID)
|
|
}
|
|
|
|
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()
|
|
defer a.mu.RUnlock()
|
|
|
|
client, ok := a.clients[clientID]
|
|
if !ok {
|
|
return false, nil
|
|
}
|
|
|
|
if client == nil {
|
|
return false, nil
|
|
}
|
|
|
|
// IsLoggedIn() checks WhatsApp session authentication, not just WebSocket connectivity.
|
|
return client.IsLoggedIn(), nil
|
|
}
|