nip47_service.go raw
1 package nip47
2
3 import (
4 "context"
5 "errors"
6 "time"
7
8 "github.com/getAlby/go-nostr"
9 "github.com/getAlby/hub/alby"
10 "github.com/getAlby/hub/apps"
11 "github.com/getAlby/hub/config"
12 "github.com/getAlby/hub/events"
13 "github.com/getAlby/hub/lnclient"
14 "github.com/getAlby/hub/logger"
15 "github.com/getAlby/hub/nip47/cipher"
16 "github.com/getAlby/hub/nip47/notifications"
17 "github.com/getAlby/hub/nip47/permissions"
18 nostrmodels "github.com/getAlby/hub/nostr/models"
19 "github.com/getAlby/hub/service/keys"
20 "github.com/getAlby/hub/transactions"
21 "github.com/sirupsen/logrus"
22 "gorm.io/gorm"
23 )
24
25 type nip47Service struct {
26 permissionsService permissions.PermissionsService
27 transactionsService transactions.TransactionsService
28 appsService apps.AppsService
29 albyOAuthSvc alby.AlbyOAuthService
30 nip47NotificationQueue notifications.Nip47NotificationQueue
31 nip47InfoPublishQueue *nip47InfoPublishQueue
32 cfg config.Config
33 keys keys.Keys
34 db *gorm.DB
35 eventPublisher events.EventPublisher
36 }
37
38 type Nip47Service interface {
39 events.EventSubscriber
40 StartNotifier(ctx context.Context, pool *nostr.SimplePool)
41 StartNip47InfoPublisher(ctx context.Context, pool *nostr.SimplePool, lnClient lnclient.LNClient)
42 HandleEvent(ctx context.Context, pool nostrmodels.SimplePool, event *nostr.Event, lnClient lnclient.LNClient)
43 GetNip47Info(ctx context.Context, pool nostrmodels.SimplePool, appWalletPubKey string) (*nostr.Event, error)
44 PublishNip47Info(ctx context.Context, pool nostrmodels.SimplePool, appId uint, appWalletPubKey string, appWalletPrivKey string, relayUrl string, lnClient lnclient.LNClient) (*nostr.Event, error)
45 PublishNip47InfoDeletion(ctx context.Context, pool nostrmodels.SimplePool, appWalletPubKey string, appWalletPrivKey string, infoEventId string) error
46 CreateResponse(initialEvent *nostr.Event, content interface{}, tags nostr.Tags, cipher *cipher.Nip47Cipher, walletPrivKey string) (result *nostr.Event, err error)
47 EnqueueNip47InfoPublishRequest(appId uint, appWalletPubKey, appWalletPrivKey, relayUrl string)
48 }
49
50 func NewNip47Service(db *gorm.DB, cfg config.Config, keys keys.Keys, eventPublisher events.EventPublisher, albyOAuthSvc alby.AlbyOAuthService) *nip47Service {
51 return &nip47Service{
52 nip47NotificationQueue: notifications.NewNip47NotificationQueue(),
53 nip47InfoPublishQueue: NewNip47InfoPublishQueue(),
54 cfg: cfg,
55 db: db,
56 permissionsService: permissions.NewPermissionsService(db, eventPublisher),
57 transactionsService: transactions.NewTransactionsService(db, eventPublisher),
58 appsService: apps.NewAppsService(db, eventPublisher, keys, cfg),
59 eventPublisher: eventPublisher,
60 keys: keys,
61 albyOAuthSvc: albyOAuthSvc,
62 }
63 }
64
65 func (svc *nip47Service) ConsumeEvent(ctx context.Context, event *events.Event, globalProperties map[string]interface{}) {
66 svc.nip47NotificationQueue.AddToQueue(event)
67 }
68
69 // The notifier is decoupled from the notification queue
70 // so that if Alby Hub disconnects from the relay, it will wait to reconnect
71 // to send notifications rather than dropping them
72 func (svc *nip47Service) StartNotifier(ctx context.Context, pool *nostr.SimplePool) {
73 nip47Notifier := notifications.NewNip47Notifier(pool, svc.db, svc.cfg, svc.keys, svc.permissionsService)
74 go func() {
75 for {
76 select {
77 case <-ctx.Done():
78 // app exited
79 return
80 case event := <-svc.nip47NotificationQueue.Channel():
81 logger.Logger.WithField("event", event).Debug("Consuming event from notification queue")
82 err := nip47Notifier.ConsumeEvent(ctx, event)
83 if err != nil {
84 logger.Logger.WithError(err).WithField("event", event).Error("Failed to consume event from notification queue")
85 // wait and then re-add the item to the queue
86 time.Sleep(5 * time.Second)
87 svc.nip47NotificationQueue.AddToQueue(event)
88 }
89 }
90 }
91 }()
92 }
93
94 func (svc *nip47Service) EnqueueNip47InfoPublishRequest(appId uint, appWalletPubKey, appWalletPrivKey, relayUrl string) {
95 svc.enqueueNip47InfoPublishRequestWithAttempt(appId, appWalletPubKey, appWalletPrivKey, relayUrl, 0)
96 }
97
98 func (svc *nip47Service) enqueueNip47InfoPublishRequestWithAttempt(appId uint, appWalletPubKey, appWalletPrivKey, relayUrl string, attempt uint32) {
99 svc.nip47InfoPublishQueue.AddToQueue(&Nip47InfoPublishRequest{
100 AppId: appId,
101 AppWalletPubKey: appWalletPubKey,
102 AppWalletPrivKey: appWalletPrivKey,
103 RelayUrl: relayUrl,
104 Attempt: attempt,
105 })
106 }
107
108 func (svc *nip47Service) StartNip47InfoPublisher(ctx context.Context, pool *nostr.SimplePool, lnClient lnclient.LNClient) {
109 go func() {
110 for {
111 select {
112 case <-ctx.Done():
113 // relay disconnected
114 return
115 case req := <-svc.nip47InfoPublishQueue.Channel():
116 _, err := svc.PublishNip47Info(ctx, pool, req.AppId, req.AppWalletPubKey, req.AppWalletPrivKey, req.RelayUrl, lnClient)
117 if err != nil {
118 // the app connection no longer exists (e.g. it was deleted),
119 // so the info event can never be published - drop the item
120 // instead of retrying forever
121 if errors.Is(err, gorm.ErrRecordNotFound) {
122 logger.Logger.WithError(err).WithFields(logrus.Fields{
123 "app_id": req.AppId,
124 "wallet_pubkey": req.AppWalletPubKey,
125 "relay_url": req.RelayUrl,
126 }).Warn("Skipping NIP47 info publish for deleted app")
127 continue
128 }
129
130 logger.Logger.WithError(err).WithFields(logrus.Fields{
131 "wallet_pubkey": req.AppWalletPubKey,
132 "relay_url": req.RelayUrl,
133 }).Error("Failed to publish NIP47 info from queue")
134
135 // wait and then re-add the item to the queue
136 // done async to ensure an offline relay does not delay
137 // the publishing of newly created app connections
138 go func() {
139 time.Sleep((5 * time.Duration(req.Attempt+1)) * time.Second)
140 svc.enqueueNip47InfoPublishRequestWithAttempt(req.AppId, req.AppWalletPubKey, req.AppWalletPrivKey, req.RelayUrl, req.Attempt+1)
141 }()
142 }
143 }
144 }
145 }()
146 }
147