package nip47 import ( "context" "errors" "time" "github.com/getAlby/go-nostr" "github.com/getAlby/hub/alby" "github.com/getAlby/hub/apps" "github.com/getAlby/hub/config" "github.com/getAlby/hub/events" "github.com/getAlby/hub/lnclient" "github.com/getAlby/hub/logger" "github.com/getAlby/hub/nip47/cipher" "github.com/getAlby/hub/nip47/notifications" "github.com/getAlby/hub/nip47/permissions" nostrmodels "github.com/getAlby/hub/nostr/models" "github.com/getAlby/hub/service/keys" "github.com/getAlby/hub/transactions" "github.com/sirupsen/logrus" "gorm.io/gorm" ) type nip47Service struct { permissionsService permissions.PermissionsService transactionsService transactions.TransactionsService appsService apps.AppsService albyOAuthSvc alby.AlbyOAuthService nip47NotificationQueue notifications.Nip47NotificationQueue nip47InfoPublishQueue *nip47InfoPublishQueue cfg config.Config keys keys.Keys db *gorm.DB eventPublisher events.EventPublisher } type Nip47Service interface { events.EventSubscriber StartNotifier(ctx context.Context, pool *nostr.SimplePool) StartNip47InfoPublisher(ctx context.Context, pool *nostr.SimplePool, lnClient lnclient.LNClient) HandleEvent(ctx context.Context, pool nostrmodels.SimplePool, event *nostr.Event, lnClient lnclient.LNClient) GetNip47Info(ctx context.Context, pool nostrmodels.SimplePool, appWalletPubKey string) (*nostr.Event, error) PublishNip47Info(ctx context.Context, pool nostrmodels.SimplePool, appId uint, appWalletPubKey string, appWalletPrivKey string, relayUrl string, lnClient lnclient.LNClient) (*nostr.Event, error) PublishNip47InfoDeletion(ctx context.Context, pool nostrmodels.SimplePool, appWalletPubKey string, appWalletPrivKey string, infoEventId string) error CreateResponse(initialEvent *nostr.Event, content interface{}, tags nostr.Tags, cipher *cipher.Nip47Cipher, walletPrivKey string) (result *nostr.Event, err error) EnqueueNip47InfoPublishRequest(appId uint, appWalletPubKey, appWalletPrivKey, relayUrl string) } func NewNip47Service(db *gorm.DB, cfg config.Config, keys keys.Keys, eventPublisher events.EventPublisher, albyOAuthSvc alby.AlbyOAuthService) *nip47Service { return &nip47Service{ nip47NotificationQueue: notifications.NewNip47NotificationQueue(), nip47InfoPublishQueue: NewNip47InfoPublishQueue(), cfg: cfg, db: db, permissionsService: permissions.NewPermissionsService(db, eventPublisher), transactionsService: transactions.NewTransactionsService(db, eventPublisher), appsService: apps.NewAppsService(db, eventPublisher, keys, cfg), eventPublisher: eventPublisher, keys: keys, albyOAuthSvc: albyOAuthSvc, } } func (svc *nip47Service) ConsumeEvent(ctx context.Context, event *events.Event, globalProperties map[string]interface{}) { svc.nip47NotificationQueue.AddToQueue(event) } // The notifier is decoupled from the notification queue // so that if Alby Hub disconnects from the relay, it will wait to reconnect // to send notifications rather than dropping them func (svc *nip47Service) StartNotifier(ctx context.Context, pool *nostr.SimplePool) { nip47Notifier := notifications.NewNip47Notifier(pool, svc.db, svc.cfg, svc.keys, svc.permissionsService) go func() { for { select { case <-ctx.Done(): // app exited return case event := <-svc.nip47NotificationQueue.Channel(): logger.Logger.WithField("event", event).Debug("Consuming event from notification queue") err := nip47Notifier.ConsumeEvent(ctx, event) if err != nil { logger.Logger.WithError(err).WithField("event", event).Error("Failed to consume event from notification queue") // wait and then re-add the item to the queue time.Sleep(5 * time.Second) svc.nip47NotificationQueue.AddToQueue(event) } } } }() } func (svc *nip47Service) EnqueueNip47InfoPublishRequest(appId uint, appWalletPubKey, appWalletPrivKey, relayUrl string) { svc.enqueueNip47InfoPublishRequestWithAttempt(appId, appWalletPubKey, appWalletPrivKey, relayUrl, 0) } func (svc *nip47Service) enqueueNip47InfoPublishRequestWithAttempt(appId uint, appWalletPubKey, appWalletPrivKey, relayUrl string, attempt uint32) { svc.nip47InfoPublishQueue.AddToQueue(&Nip47InfoPublishRequest{ AppId: appId, AppWalletPubKey: appWalletPubKey, AppWalletPrivKey: appWalletPrivKey, RelayUrl: relayUrl, Attempt: attempt, }) } func (svc *nip47Service) StartNip47InfoPublisher(ctx context.Context, pool *nostr.SimplePool, lnClient lnclient.LNClient) { go func() { for { select { case <-ctx.Done(): // relay disconnected return case req := <-svc.nip47InfoPublishQueue.Channel(): _, err := svc.PublishNip47Info(ctx, pool, req.AppId, req.AppWalletPubKey, req.AppWalletPrivKey, req.RelayUrl, lnClient) if err != nil { // the app connection no longer exists (e.g. it was deleted), // so the info event can never be published - drop the item // instead of retrying forever if errors.Is(err, gorm.ErrRecordNotFound) { logger.Logger.WithError(err).WithFields(logrus.Fields{ "app_id": req.AppId, "wallet_pubkey": req.AppWalletPubKey, "relay_url": req.RelayUrl, }).Warn("Skipping NIP47 info publish for deleted app") continue } logger.Logger.WithError(err).WithFields(logrus.Fields{ "wallet_pubkey": req.AppWalletPubKey, "relay_url": req.RelayUrl, }).Error("Failed to publish NIP47 info from queue") // wait and then re-add the item to the queue // done async to ensure an offline relay does not delay // the publishing of newly created app connections go func() { time.Sleep((5 * time.Duration(req.Attempt+1)) * time.Second) svc.enqueueNip47InfoPublishRequestWithAttempt(req.AppId, req.AppWalletPubKey, req.AppWalletPrivKey, req.RelayUrl, req.Attempt+1) }() } } } }() }