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