Repository navigation
Expand file tree
/
Copy pathwmiau.go
More file actions
1850 lines (1674 loc) · 59 KB
/
Copy pathwmiau.go
File metadata and controls
1850 lines (1674 loc) · 59 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
package main
import (
"context"
"crypto/tls"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"mime"
"net/http"
"net/url"
"os"
"path/filepath"
"strconv"
"strings"
"time"
"github.com/go-resty/resty/v2"
"github.com/jmoiron/sqlx"
"github.com/mdp/qrterminal/v3"
"github.com/patrickmn/go-cache"
"github.com/rs/zerolog/log"
"github.com/skip2/go-qrcode"
"go.mau.fi/whatsmeow"
"go.mau.fi/whatsmeow/appstate"
"go.mau.fi/whatsmeow/proto/waCompanionReg"
"go.mau.fi/whatsmeow/store"
"go.mau.fi/whatsmeow/types"
"go.mau.fi/whatsmeow/types/events"
waLog "go.mau.fi/whatsmeow/util/log"
"golang.org/x/net/proxy"
)
// db field declaration as *sqlx.DB
type MyClient struct {
WAClient *whatsmeow.Client
eventHandlerID uint32
userID string
token string
subscriptions []string
db *sqlx.DB
s *server
}
func sendToGlobalWebHook(jsonData []byte, token string, userID string) {
jsonDataStr := string(jsonData)
instance_name := ""
userinfo, found := userinfocache.Get(token)
if found {
instance_name = userinfo.(Values).Get("Name")
}
if *globalWebhook != "" {
log.Info().Str("url", *globalWebhook).Msg("Calling global webhook")
// Add extra information for the global webhook
globalData := map[string]string{
"jsonData": jsonDataStr,
"userID": userID,
"instanceName": instance_name,
}
callHookWithHmac(*globalWebhook, globalData, userID, globalHMACKeyEncrypted)
}
}
func sendToUserWebHook(webhookurl string, path string, jsonData []byte, userID string, token string) {
sendToUserWebHookWithHmac(webhookurl, path, jsonData, userID, token, nil)
}
func sendToUserWebHookWithHmac(webhookurl string, path string, jsonData []byte, userID string, token string, encryptedHmacKey []byte) {
instance_name := ""
userinfo, found := userinfocache.Get(token)
if found {
instance_name = userinfo.(Values).Get("Name")
}
data := map[string]string{
"jsonData": string(jsonData),
"userID": userID,
"instanceName": instance_name,
}
log.Debug().Interface("webhookData", data).Msg("Data being sent to webhook")
if webhookurl != "" {
log.Info().Str("url", webhookurl).Msg("Calling user webhook")
if path == "" {
go callHookWithHmac(webhookurl, data, userID, encryptedHmacKey)
} else {
// Create a channel to capture the error from the goroutine
errChan := make(chan error, 1)
go func() {
err := callHookFileWithHmac(webhookurl, data, userID, path, encryptedHmacKey)
errChan <- err
}()
// Optionally handle the error from the channel (if needed)
if err := <-errChan; err != nil {
log.Error().Err(err).Msg("Error calling hook file")
}
}
} else {
log.Warn().Str("userid", userID).Msg("No webhook se
8000
t for user")
}
}
func updateAndGetUserSubscriptions(mycli *MyClient) ([]string, error) {
// Get updated events from cache/database
currentEvents := ""
userinfo2, found2 := userinfocache.Get(mycli.token)
if found2 {
currentEvents = userinfo2.(Values).Get("Events")
} else {
// If not in cache, get from database
if err := mycli.db.Get(¤tEvents, "SELECT events FROM users WHERE id=$1", mycli.userID); err != nil {
log.Warn().Err(err).Str("userID", mycli.userID).Msg("Could not get events from DB")
return nil, err // Propagate the error
}
}
// Update client subscriptions if changed
eventarray := strings.Split(currentEvents, ",")
var subscribedEvents []string
if len(eventarray) == 1 && eventarray[0] == "" {
subscribedEvents = []string{}
} else {
for _, arg := range eventarray {
arg = strings.TrimSpace(arg)
if arg != "" && Find(supportedEventTypes, arg) {
subscribedEvents = append(subscribedEvents, arg)
}
}
}
// Update the client subscriptions
mycli.subscriptions = subscribedEvents
return subscribedEvents, nil
}
func getUserWebhookUrl(token string) string {
webhookurl := ""
myuserinfo, found := userinfocache.Get(token)
if !found {
log.Warn().Str("token", token).Msg("Could not call webhook as there is no user for this token")
} else {
webhookurl = myuserinfo.(Values).Get("Webhook")
}
return webhookurl
}
func sendEventWithWebHook(mycli *MyClient, postmap map[string]interface{}, path string) {
webhookurl := getUserWebhookUrl(mycli.token)
// Get updated events from cache/database
subscribedEvents, err := updateAndGetUserSubscriptions(mycli)
if err != nil {
return
}
eventType, ok := postmap["type"].(string)
if !ok {
log.Error().Msg("Event type is not a string in postmap")
return
}
// Log subscription details for debugging
log.Debug().
Str("userID", mycli.userID).
Str("eventType", eventType).
Strs("subscribedEvents", subscribedEvents).
Msg("Checking event subscription")
// Check if the current event is in the subscriptions
checkIfSubscribedInEvent := checkIfSubscribedToEvent(subscribedEvents, postmap["type"].(string), mycli.userID)
if !checkIfSubscribedInEvent {
return
}
// In stdio mode, send as JSON-RPC notification instead of HTTP webhook
if mycli.s != nil && mycli.s.mode == Stdio {
mycli.s.SendNotification(eventType, postmap)
return
}
// Prepare webhook data
jsonData, err := json.Marshal(postmap)
if err != nil {
log.Error().Err(err).Msg("Failed to marshal postmap to JSON")
return
}
// Get HMAC key for this user
var encryptedHmacKey []byte
if userinfo, found := userinfocache.Get(mycli.token); found {
encryptedB64 := userinfo.(Values).Get("HmacKeyEncrypted")
if encryptedB64 != "" {
var err error
encryptedHmacKey, err = base64.StdEncoding.DecodeString(encryptedB64)
if err != nil {
log.Error().Err(err).Msg("Failed to decode HMAC key from cache")
}
}
}
sendToUserWebHookWithHmac(webhookurl, path, jsonData, mycli.userID, mycli.token, encryptedHmacKey)
// Get global webhook if configured
go sendToGlobalWebHook(jsonData, mycli.token, mycli.userID)
go sendToGlobalRabbit(jsonData, mycli.token, mycli.userID)
}
func checkIfSubscribedToEvent(subscribedEvents []string, eventType string, userId string) bool {
if !Find(subscribedEvents, eventType) && !Find(subscribedEvents, "All") {
log.Warn().
Str("type", eventType).
Strs("subscribedEvents", subscribedEvents).
Str("userID", userId).
Msg("Skipping webhook. Not subscribed for this type")
return false
}
return true
}
// Connects to Whatsapp Websocket on server startup if last state was connected
func (s *server) connectOnStartup() {
rows, err := s.db.Queryx("SELECT id,name,token,jid,webhook,events,proxy_url,CASE WHEN s3_enabled THEN 'true' ELSE 'false' END AS s3_enabled,media_delivery,COALESCE(history, 0) as history,hmac_key FROM users WHERE connected=1")
if err != nil {
log.Error().Err(err).Msg("DB Problem")
return
}
defer rows.Close()
for rows.Next() {
txtid := ""
token := ""
jid := ""
name := ""
webhook := ""
events := ""
proxy_url := ""
s3_enabled := ""
media_delivery := ""
var history int
var hmac_key []byte
err = rows.Scan(&txtid, &name, &token, &jid, &webhook, &events, &proxy_url, &s3_enabled, &media_delivery, &history, &hmac_key)
if err != nil {
log.Error().Err(err).Msg("DB Problem")
return
} else {
hmacKeyEncrypted := ""
if len(hmac_key) > 0 {
hmacKeyEncrypted = base64.StdEncoding.EncodeToString(hmac_key)
}
log.Info().Str("token", token).Msg("Connect to Whatsapp on startup")
v := Values{map[string]string{
"Id": txtid,
"Name": name,
"Jid": jid,
"Webhook": webhook,
"Token": token,
"Proxy": proxy_url,
"Events": events,
"S3Enabled": s3_enabled,
"MediaDelivery": media_delivery,
"History": fmt.Sprintf("%d", history),
"HmacKeyEncrypted": hmacKeyEncrypted,
}}
userinfocache.Set(token, v, cache.NoExpiration)
// Gets and set subscription to webhook events
eventarray := strings.Split(events, ",")
var subscribedEvents []string
if len(eventarray) == 1 && eventarray[0] == "" {
subscribedEvents = []string{}
} else {
for _, arg := range eventarray {
if !Find(supportedEventTypes, arg) {
log.Warn().Str("Type", arg).Msg("Event type discarded")
continue
}
if !Find(subscribedEvents, arg) {
subscribedEvents = append(subscribedEvents, arg)
}
}
}
eventstring := strings.Join(subscribedEvents, ",")
log.Info().Str("events", eventstring).Str("jid", jid).Msg("Attempt to connect")
killchannel[txtid] = make(chan bool)
go s.startClient(txtid, jid, token, subscribedEvents)
// Initialize S3 client if configured
go func(userID string) {
var s3Config struct {
Enabled bool `db:"s3_enabled"`
Endpoint string `db:"s3_endpoint"`
Region string `db:"s3_region"`
Bucket string `db:"s3_bucket"`
AccessKey string `db:"s3_access_key"`
SecretKey string `db:"s3_secret_key"`
PathStyle bool `db:"s3_path_style"`
PublicURL string `db:"s3_public_url"`
RetentionDays int `db:"s3_retention_days"`
}
err := s.db.Get(&s3Config, `
SELECT s3_enabled, s3_endpoint, s3_region, s3_bucket,
s3_access_key, s3_secret_key, s3_path_style,
s3_public_url, s3_retention_days
FROM users WHERE id = $1`, userID)
if err != nil {
log.Error().Err(err).Str("userID", userID).Msg("Failed to get S3 config")
return
}
if s3Config.Enabled {
config := &S3Config{
Enabled: s3Config.Enabled,
Endpoint: s3Config.Endpoint,
Region: s3Config.Region,
Bucket: s3Config.Bucket,
AccessKey: s3Config.AccessKey,
SecretKey: s3Config.SecretKey,
PathStyle: s3Config.PathStyle,
PublicURL: s3Config.PublicURL,
RetentionDays: s3Config.RetentionDays,
}
err = GetS3Manager().InitializeS3Client(userID, config)
if err != nil {
log.Error().Err(err).Str("userID", userID).Msg("Failed to initialize S3 client on startup")
} else {
log.Info().Str("userID", userID).Msg("S3 client initialized on startup")
}
}
}(txtid)
}
}
err = rows.Err()
if err != nil {
log.Error().Err(err).Msg("DB Problem")
}
}
func parseJID(arg string) (types.JID, bool) {
if arg[0] == '+' {
arg = arg[1:]
}
if !strings.ContainsRune(arg, '@') {
return types.NewJID(arg, types.DefaultUserServer), true
} else {
recipient, err := types.ParseJID(arg)
if err != nil {
log.Error().Err(err).Msg("Invalid JID")
return recipient, false
} else if recipient.User == "" {
log.Error().Err(err).Msg("Invalid JID no server specified")
return recipient, false
}
return recipient, true
}
}
func (s *server) startClient(userID string, textjid string, token string, subscriptions []string) {
log.Info().Str("userid", userID).Str("jid", textjid).Msg("Starting websocket connection to Whatsapp")
// Connection retry constants
const maxConnectionRetries = 3
const connectionRetryBaseWait = 5 * time.Second
var deviceStore *store.Device
var err error
// First handle the device store initialization
if textjid != "" {
jid, _ := parseJID(textjid)
deviceStore, err = container.GetDevice(context.Background(), jid)
if err != nil {
log.Error().Err(err).Msg("Failed to get device")
deviceStore = container.NewDevice()
}
} else {
log.Warn().Msg("No jid found. Creating new device")
deviceStore = container.NewDevice()
}
if deviceStore == nil {
log.Warn().Msg("No store found. Creating new one")
deviceStore = container.NewDevice()
}
clientLog := waLog.Stdout("Client", *waDebug, *colorOutput)
// Create the client with initialized deviceStore
var client *whatsmeow.Client
if *waDebug != "" {
client = whatsmeow.NewClient(deviceStore, clientLog)
} else {
client = whatsmeow.NewClient(deviceStore, nil)
}
// Now we can use the client with the manager
clientManager.SetWhatsmeowClient(userID, client)
store.DeviceProps.PlatformType = waCompanionReg.DeviceProps_DESKTOP.Enum()
store.DeviceProps.Os = osName
mycli := MyClient{client, 1, userID, token, subscriptions, s.db, s}
mycli.eventHandlerID = mycli.WAClient.AddEventHandler(mycli.myEventHandler)
// Store the MyClient in clientManager
clientManager.SetMyClient(userID, &mycli)
httpClient := resty.New()
httpClient.SetRedirectPolicy(resty.FlexibleRedirectPolicy(15))
if *waDebug == "DEBUG" {
httpClient.SetDebug(true)
}
httpClient.SetTimeout(30 * time.Second)
httpClient.SetTLSClientConfig(&tls.Config{InsecureSkipVerify: true})
httpClient.OnError(func(req *resty.Request, err error) {
if v, ok := err.(*resty.ResponseError); ok {
// v.Response contains the last response from the server
// v.Err contains the original error
log.Debug().Str("response", v.Response.String()).Msg("resty error")
log.Error().Err(v.Err).Msg("resty error")
}
})
// Set proxy if defined in DB (assumes users table contains proxy_url column)
var proxyURL string
err = s.db.Get(&proxyURL, "SELECT proxy_url FROM users WHERE id=$1", userID)
if err == nil && proxyURL != "" {
parsed, perr := url.Parse(proxyURL)
if perr != nil {
log.Warn().Err(perr).Str("proxy", proxyURL).Msg("Invalid proxy URL, skipping proxy setup")
} else {
log.Info().Str("proxy", proxyURL).Msg("Configuring proxy")
if parsed.Scheme == "socks5" || parsed.Scheme == "socks5h" {
dialer, derr := proxy.FromURL(parsed, nil)
if derr != nil {
log.Warn().Err(derr).Str("proxy", proxyURL).Msg("Failed to build SOCKS proxy dialer, skipping proxy setup")
} else {
httpClient.SetProxy(proxyURL)
client.SetSOCKSProxy(dialer, whatsmeow.SetProxyOptions{})
log.Info().Msg("SOCKS proxy configured successfully")
}
} else {
httpClient.SetProxy(proxyURL)
client.SetProxyAddress(parsed.String(), whatsmeow.SetProxyOptions{})
log.Info().Msg("HTTP/HTTPS proxy configured successfully")
}
}
}
clientManager.SetHTTPClient(userID, httpClient)
if client.Store.ID == nil {
// No ID stored, new login
qrChan, err := client.GetQRChannel(context.Background())
if err != nil {
// This error means that we're already logged in, so ignore it.
if !errors.Is(err, whatsmeow.ErrQRStoreContainsID) {
log.Error().Err(err).Msg("Failed to get QR channel")
return
}
} else {
err = client.Connect() // Must connect to generate QR code
if err != nil {
log.Error().Err(err).Msg("Failed to connect client")
return
}
myuserinfo, found := userinfocache.Get(token)
for evt := range qrChan {
if evt.Event == "code" {
// Display QR code in terminal (useful for testing/developing)
// Skip in stdio mode to avoid breaking JSON-RPC
if *logType != "json" && s.mode != Stdio {
qrterminal.GenerateHalfBlock(evt.Code, qrterminal.L, os.Stdout)
fmt.Println("QR code:\n", evt.Code)
}
// Store encoded/embeded base64 QR on database for retrieval with the /qr endpoint
image, _ := qrcode.Encode(evt.Code, qrcode.Medium, 256)
base64qrcode := "data:image/png;base64," + base64.StdEncoding.EncodeToString(image)
sqlStmt := `UPDATE users SET qrcode=$1 WHERE id=$2`
_, err := s.db.Exec(sqlStmt, base64qrcode, userID)
if err != nil {
log.Error().Err(err).Msg(sqlStmt)
} else {
if found {
v := updateUserInfo(myuserinfo, "Qrcode", base64qrcode)
userinfocache.Set(token, v, cache.NoExpiration)
log.Info().Str("qrcode", base64qrcode).Msg("update cache userinfo with qr code")
}
}
//send QR code with webhook
postmap := make(map[string]interface{})
postmap["event"] = evt.Event
postmap["qrCodeBase64"] = base64qrcode
postmap["type"] = "QR"
sendEventWithWebHook(&mycli, postmap, "")
} else if evt.Event == "timeout" {
// Clear QR code from DB on timeout
// Send webhook notifying QR timeout before cleanup
postmap := make(map[string]interface{})
postmap["event"] = evt.Event
postmap["type"] = "QRTimeout"
sendEventWithWebHook(&mycli, postmap, "")
sqlStmt := `UPDATE users SET qrcode='' WHERE id=$1`
_, err := s.db.Exec(sqlStmt, userID)
if err != nil {
log.Error().Err(err).Msg(sqlStmt)
} else {
if found {
v := updateUserInfo(myuserinfo, "Qrcode", "")
userinfocache.Set(token, v, cache.NoExpiration)
}
}
log.Warn().Msg("QR timeout killing channel")
clientManager.DeleteWhatsmeowClient(userID)
clientManager.DeleteMyClient(userID)
clientManager.DeleteHTTPClient(userID)
killchannel[userID] <- true
} else if evt.Event == "success" {
log.Info().Msg("QR pairing ok!")
// Clear QR code after pairing
sqlStmt := `UPDATE users SET qrcode='', connected=1 WHERE id=$1`
_, err := s.db.Exec(sqlStmt, userID)
if err != nil {
log.Error().Err(err).Msg(sqlStmt)
} else {
if found {
v := updateUserInfo(myuserinfo, "Qrcode", "")
userinfocache.Set(token, v, cache.NoExpiration)
}
}
} else {
log.Info().Str("event", evt.Event).Msg("Login event")
}
}
}
} else {
// Already logged in, just connect
log.Info().Msg("Already logged in, just connect")
// Retry logic with linear backoff
var lastErr error
for attempt := 0; attempt < maxConnectionRetries; attempt++ {
if attempt > 0 {
waitTime := time.Duration(attempt) * connectionRetryBaseWait
log.Warn().
Int("attempt", attempt+1).
Int("max_retries", maxConnectionRetries).
Dur("wait_time", waitTime).
Msg("Retrying connection after delay")
time.Sleep(waitTime)
}
err = client.Connect()
if err == nil {
log.Info().
Int("attempt", attempt+1).
Msg("Successfully connected to WhatsApp")
break
}
lastErr = err
log.Warn().
Err(err).
Int("attempt", attempt+1).
Int("max_retries", maxConnectionRetries).
Msg("Failed to connect to WhatsApp")
}
if lastErr != nil {
log.Error().
Err(lastErr).
Str("userid", userID).
Int("attempts", maxConnectionRetries).
Msg("Failed to connect to WhatsApp after all retry attempts")
clientManager.DeleteWhatsmeowClient(userID)
clientManager.DeleteMyClient(userID)
clientManager.DeleteHTTPClient(userID)
sqlStmt := `UPDATE users SET qrcode='', connected=0 WHERE id=$1`
_, dbErr := s.db.Exec(sqlStmt, userID)
if dbErr != nil {
log.Error().Err(dbErr).Msg("Failed to update user status after connection error")
}
// Use the existing mycli instance from outer scope
postmap := make(map[string]interface{})
postmap["event"] = "ConnectFailure"
postmap["error"] = lastErr.Error()
postmap["type"] = "ConnectFailure"
postmap["attempts"] = maxConnectionRetries
postmap["reason"] = "Failed to connect after retry attempts"
sendEventWithWebHook(&mycli, postmap, "")
return
}
}
// Keep connected client live until disconnected/killed
for {
select {
case <-killchannel[userID]:
log.Info().Str("userid", userID).Msg("Received kill signal")
client.Disconnect()
clientManager.DeleteWhatsmeowClient(userID)
clientManager.DeleteMyClient(userID)
clientManager.DeleteHTTPClient(userID)
sqlStmt := `UPDATE users SET qrcode='', connected=0 WHERE id=$1`
_, err := s.db.Exec(sqlStmt, userID)
if err != nil {
log.Error().Err(err).Msg(sqlStmt)
}
return
default:
time.Sleep(1000 * time.Millisecond)
//log.Info().Str("jid",textjid).Msg("Loop the loop")
}
}
}
func fileToBase64(filepath string) (string, string, error) {
data, err := os.ReadFile(filepath)
if err != nil {
return "", "", err
}
mimeType := http.DetectContentType(data)
return base64.StdEncoding.EncodeToString(data), mimeType, nil
}
func (mycli *MyClient) myEventHandler(rawEvt interface{}) {
txtid := mycli.userID
postmap := make(map[string]interface{})
postmap["event"] = rawEvt
dowebhook := 0
path := ""
switch evt := rawEvt.(type) {
case *events.AppStateSyncComplete:
if len(mycli.WAClient.Store.PushName) > 0 && evt.Name == appstate.WAPatchCriticalBlock {
err := mycli.WAClient.SendPresence(context.Background(), types.PresenceAvailable)
if err != nil {
log.Warn().Err(err).Msg("Failed to send available presence")
} else {
log.Info().Msg("Marked self as available")
}
}
case *events.Connected, *events.PushNameSetting:
postmap["type"] = "Connected"
dowebhook = 1
if len(mycli.WAClient.Store.PushName) == 0 {
break
}
// Send presence available when connecting and when the pushname is changed.
// This makes sure that outgoing messages always have the right pushname.
err := mycli.WAClient.SendPresence(context.Background(), types.PresenceAvailable)
if err != nil {
log.Warn().Err(err).Msg("Failed to send available presence")
} else {
log.Info().Msg("Marked self as available")
}
sqlStmt := `UPDATE users SET connected=1 WHERE id=$1`
_, err = mycli.db.Exec(sqlStmt, mycli.userID)
if err != nil {
log.Error().Err(err).Msg(sqlStmt)
return
}
case *events.PairSuccess:
log.Info().Str("userid", mycli.userID).Str("token", mycli.token).Str("ID", evt.ID.String()).Str("BusinessName", evt.BusinessName).Str("Platform", evt.Platform).Msg("QR Pair Success")
jid := evt.ID
sqlStmt := `UPDATE users SET jid=$1 WHERE id=$2`
_, err := mycli.db.Exec(sqlStmt, jid, mycli.userID)
if err != nil {
log.Error().Err(err).Msg(sqlStmt)
return
}
postmap["type"] = "PairSuccess"
dowebhook = 1
myuserinfo, found := userinfocache.Get(mycli.token)
if !found {
log.Warn().Msg("No user info cached on pairing?")
} else {
txtid = myuserinfo.(Values).Get("Id")
token := myuserinfo.(Values).Get("Token")
v := updateUserInfo(myuserinfo, "Jid", fmt.Sprintf("%s", jid))
userinfocache.Set(token, v, cache.NoExpiration)
log.Info().Str("jid", jid.String()).Str("userid", txtid).Str("token", token).Msg("User information set")
}
// Check if automatic history sync is enabled and trigger it after QR code is scanned
var daysToSyncHistory int
query := "SELECT COALESCE(days_to_sync_history, 0) FROM users WHERE id=$1"
query = mycli.db.Rebind(query)
err = mycli.db.Get(&daysToSyncHistory, query, mycli.userID)
if err != nil {
log.Warn().Err(err).Str("userID", mycli.userID).Msg("Failed to get days_to_sync_history from database")
} else if daysToSyncHistory > 0 {
// Trigger history sync in a goroutine to avoid blocking
// Wait a bit for the connection to be fully established
go func() {
time.Sleep(2 * time.Second) // Give WhatsApp time to fully establish connection
log.Info().
Str("userID", mycli.userID).
Int("days", daysToSyncHistory).
Msg("Triggering automatic history sync after QR code scan")
// Use the SyncWhatsAppHistory logic but for a single user
// Calculate message count based on days (estimate: 15 messages per day)
count := daysToSyncHistory * 15
if count > 500 {
count = 500 // WhatsApp limit
}
if count < 50 {
count = 50 // Minimum reasonable count
}
// Get chats from WhatsApp (contacts and groups)
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
var chatJIDs []string
// Get all contacts
contacts, err := mycli.WAClient.Store.Contacts.GetAllContacts(ctx)
if err != nil {
log.Error().Err(err).Str("userID", mycli.userID).Msg("Failed to get contacts for history sync")
} else {
for jid := range contacts {
chatJIDs = append(chatJIDs, jid.String())
}
}
// Get all groups
groups, err := mycli.WAClient.GetJoinedGroups(ctx)
if err != nil {
log.Error().Err(err).Str("userID", mycli.userID).Msg("Failed to get groups for history sync")
} else {
for _, group := range groups {
chatJIDs = append(chatJIDs, group.JID.String())
}
}
// Sync each chat with a small delay between requests
for _, chatJIDStr := range chatJIDs {
chatJID, err := types.ParseJID(chatJIDStr)
if err != nil {
log.Warn().Err(err).Str("chatJID", chatJIDStr).Msg("Failed to parse chat JID, skipping")
continue
}
// Use the syncHistoryForChat function from handlers.go
err = mycli.s.syncHistoryForChat(context.Background(), mycli.userID, chatJID, count)
if err != nil {
log.Warn().Err(err).Str("chatJID", chatJIDStr).Msg("Failed to sync history for chat")
} else {
log.Info().Str("chatJID", chatJIDStr).Int("count", count).Msg("History sync request sent for chat")
}
// Small delay between requests to avoid overwhelming WhatsApp
time.Sleep(100 * time.Millisecond)
}
log.Info().
Str("userID", mycli.userID).
Int("days", daysToSyncHistory).
Int("chatsSynced", len(chatJIDs)).
Msg("Automatic history sync completed after QR code scan")
}()
}
case *events.StreamReplaced:
log.Info().Msg("Received StreamReplaced event")
return
case *events.Message:
var s3Config struct {
Enabled string `db:"s3_enabled"`
MediaDelivery string `db:"media_delivery"`
}
lastMessageCache.Set(mycli.userID, &evt.Info, cache.DefaultExpiration)
myuserinfo, found := userinfocache.Get(mycli.token)
if !found {
err := mycli.db.Get(&s3Config, "SELECT CASE WHEN s3_enabled = 1 THEN 'true' ELSE 'false' END AS s3_enabled, media_delivery FROM users WHERE id = $1", txtid)
if err != nil {
log.Error().Err(err).Msg("onMessage Failed to get S3 config from DB as it was not on cache")
s3Config.Enabled = "false"
s3Config.MediaDelivery = "base64"
}
} else {
s3Config.Enabled = myuserinfo.(Values).Get("S3Enabled")
s3Config.MediaDelivery = myuserinfo.(Values).Get("MediaDelivery")
}
postmap["type"] = "Message"
dowebhook = 1
metaParts := []string{fmt.Sprintf("pushname: %s", evt.Info.PushName), fmt.Sprintf("timestamp: %s", evt.Info.Timestamp)}
if evt.Info.Type != "" {
metaParts = append(metaParts, fmt.Sprintf("type: %s", evt.Info.Type))
}
if evt.Info.Category != "" {
metaParts = append(metaParts, fmt.Sprintf("category: %s", evt.Info.Category))
}
if evt.IsViewOnce {
metaParts = append(metaParts, "view once")
}
if evt.IsViewOnce {
metaParts = append(metaParts, "ephemeral")
}
log.Info().Str("id", evt.Info.ID).Str("source", evt.Info.SourceString()).Str("parts", strings.Join(metaParts, ", ")).Msg("Message Received")
if !*skipMedia {
// try to get Image if any
img := evt.Message.GetImageMessage()
if img != nil {
// Create a temporary directory in /tmp
tmpDirectory := filepath.Join("/tmp", "user_"+txtid)
errDir := os.MkdirAll(tmpDirectory, 0751)
if errDir != nil {
log.Error().Err(errDir).Msg("Could not create temporary directory")
return
}
// Download the image
data, err := mycli.WAClient.Download(context.Background(), img)
if err != nil {
log.Error().Err(err).Msg("Failed to download image")
return
}
// Determine the file extension based on the MIME type
exts, _ := mime.ExtensionsByType(img.GetMimetype())
tmpPath := filepath.Join(tmpDirectory, evt.Info.ID+exts[0])
// Write the image to the temporary file
err = os.WriteFile(tmpPath, data, 0600)
if err != nil {
log.Error().Err(err).Msg("Failed to save image to temporary file")
return
}
// Process S3 upload if enabled
if s3Config.Enabled == "true" && (s3Config.MediaDelivery == "s3" || s3Config.MediaDelivery == "both") {
// Get sender JID for inbox/outbox determination
isIncoming := evt.Info.IsFromMe == false
contactJID := evt.Info.Sender.String()
if evt.Info.IsGroup {
contactJID = evt.Info.Chat.String()
}
// Process S3 upload
s3Data, err := GetS3Manager().ProcessMediaForS3(
context.Background(),
txtid,
contactJID,
evt.Info.ID,
data,
img.GetMimetype(),
filepath.Base(tmpPath),
isIncoming,
)
if err != nil {
log.Error().Err(err).Msg("Failed to upload image to S3")
} else {
postmap["s3"] = s3Data
}
}
// Convert the image to base64 if needed
if s3Config.MediaDelivery == "base64" || s3Config.MediaDelivery == "both" {
base64String, mimeType, err := fileToBase64(tmpPath)
if err != nil {
log.Error().Err(err).Msg("Failed to convert image to base64")
return
}
// Add the base64 string and other details to the postmap
postmap["base64"] = base64String
postmap["mimeType"] = mimeType
postmap["fileName"] = filepath.Base(tmpPath)
}
// Log the successful conversion
log.Info().Str("path", tmpPath).Msg("Image processed")
// Delete the temporary file
err = os.Remove(tmpPath)
if err != nil {
log.Error().Err(err).Msg("Failed to delete temporary file")
} else {
log.Info().Str("path", tmpPath).Msg("Temporary file deleted")
}
}
// try to get Audio if any
audio := evt.Message.GetAudioMessage()
if audio != nil {
// Create a temporary directory in /tmp
tmpDirectory := filepath.Join("/tmp", "user_"+txtid)
errDir := os.MkdirAll(tmpDirectory, 0751)
if errDir != nil {
log.Error().Err(errDir).Msg("Could not create temporary directory")
return
}
// Download the audio
data, err := mycli.WAClient.Download(context.Background(), audio)
if err != nil {
log.Error().Err(err).Msg("Failed to download audio")
return
}
// Determine the file extension based on the MIME type
exts, _ := mime.ExtensionsByType(audio.GetMimetype())
var ext string
if len(exts) > 0 {
ext = exts[0]
} else {
ext = ".ogg" // Default extension if MIME type is not recognized
}
tmpPath := filepath.Join(tmpDirectory, evt.Info.ID+ext)
// Write the audio to the temporary file
err = os.WriteFile(tmpPath, data, 0600)
if err != nil {
log.Error().Err(err).Msg("Failed to save audio to temporary file")
return
}
// Process S3 upload if enabled
if s3Config.Enabled == "true" && (s3Config.MediaDelivery == "s3" || s3Config.MediaDelivery == "both") {
// Get sender JID for inbox/outbox determination
isIncoming := evt.Info.IsFromMe == false
contactJID := evt.Info.Sender.String()
if evt.Info.IsGroup {
contactJID = evt.Info.Chat.String()
}
// Process S3 upload
s3Data, err := GetS3Manager().ProcessMediaForS3(
context.Background(),
txtid,
contactJID,
evt.Info.ID,
data,
audio.GetMimetype(),
filepath.Base(tmpPath),
isIncoming,
)
if err != nil {
log.Error().Err(err).Msg("Failed to upload audio to S3")
} else {
postmap["s3"] = s3Data
}
}
// Convert the audio to base64 if needed
if s3Config.MediaDelivery == "base64" || s3Config.MediaDelivery == "both" {
base64String, mimeType, err := fileToBase64(tmpPath)
if err != nil {
log.Error().Err(err).Msg("Failed to convert audio to base64")
return
}
// Add the base64 string and other details to the postmap
postmap["base64"] = base64String
postmap["mimeType"] = mimeType
postmap["fileName"] = filepath.Base(tmpPath)
}
// Log the successful conversion
log.Info().Str("path", tmpPath).Msg("Audio processed")
// Delete the temporary file
err = os.Remove(tmpPath)
if err != nil {
log.Error().Err(err).Msg("Failed to delete temporary file")
} else {