diff --git a/internal/nostr/models.go b/internal/nostr/models.go index 756f3b6..7ca9f02 100644 --- a/internal/nostr/models.go +++ b/internal/nostr/models.go @@ -25,30 +25,31 @@ const ( ) type Subscription struct { - ID uint - RelayUrl string - WebhookUrl string - PushToken string - IsIOS bool - Open bool - Ids *[]string `gorm:"-"` - Kinds *[]int `gorm:"-"` - Authors *[]string `gorm:"-"` // WalletPubkey is included in this - Tags *nostr.TagMap `gorm:"-"` // RequestEvent ID goes in the "e" tag - Since time.Time - Until time.Time - Limit int - Search string - CreatedAt time.Time - UpdatedAt time.Time - Uuid string `gorm:"type:uuid;default:gen_random_uuid()"` - EventChan chan *nostr.Event `gorm:"-"` - RequestEvent *RequestEvent `gorm:"-"` - - IdsJson json.RawMessage `gorm:"type:jsonb"` - KindsJson json.RawMessage `gorm:"type:jsonb"` - AuthorsJson json.RawMessage `gorm:"type:jsonb"` - TagsJson json.RawMessage `gorm:"type:jsonb"` + ID uint + RelayUrl string + WebhookUrl string + PushToken string + IsIOS bool + Open bool + Ids *[]string `gorm:"-"` + Kinds *[]int `gorm:"-"` + Authors *[]string `gorm:"-"` // WalletPubkey is included in this + Tags *nostr.TagMap `gorm:"-"` // RequestEvent ID goes in the "e" tag + Since time.Time + Until time.Time + Limit int + Search string + CreatedAt time.Time + UpdatedAt time.Time + Uuid string `gorm:"type:uuid;default:gen_random_uuid()"` + EventChan chan *nostr.Event `gorm:"-"` + RequestEvent *RequestEvent `gorm:"-"` + RelaySubscription *nostr.Subscription `gorm:"-"` + + IdsJson json.RawMessage `gorm:"type:jsonb"` + KindsJson json.RawMessage `gorm:"type:jsonb"` + AuthorsJson json.RawMessage `gorm:"type:jsonb"` + TagsJson json.RawMessage `gorm:"type:jsonb"` } func (s *Subscription) BeforeSave(tx *gorm.DB) error { diff --git a/internal/nostr/nostr.go b/internal/nostr/nostr.go index 9661f75..2690910 100644 --- a/internal/nostr/nostr.go +++ b/internal/nostr/nostr.go @@ -47,10 +47,10 @@ type Service struct { Relay *nostr.Relay Cfg *Config Logger *logrus.Logger - subscriptions map[string]*nostr.Subscription subscriptionsMutex sync.Mutex relayMutex sync.Mutex client *expo.PushClient + subCancelFnMap map[string]context.CancelFunc } func NewService(ctx context.Context) (*Service, error) { @@ -116,8 +116,6 @@ func NewService(ctx context.Context) (*Service, error) { return nil, err } - subscriptions := make(map[string]*nostr.Subscription) - client := expo.NewPushClient(&expo.ClientConfig{ Host: "https://api.expo.dev", APIURL: "/v2", @@ -131,7 +129,6 @@ func NewService(ctx context.Context) (*Service, error) { Wg: &wg, Logger: logger, Relay: relay, - subscriptions: subscriptions, client: client, } @@ -142,7 +139,7 @@ func NewService(ctx context.Context) (*Service, error) { logger.WithError(err).Error("Failed to query open subscriptions") return nil, err } - + cancelFnMap := make(map[string]context.CancelFunc) for _, sub := range openSubscriptions { // Create a copy of the loop variable to // avoid passing address of the same variable @@ -151,8 +148,11 @@ func NewService(ctx context.Context) (*Service, error) { if sub.PushToken != "" { handleEvent = svc.handleSubscribedEventForPushNotification } - go svc.startSubscription(svc.Ctx, &subscription, nil, handleEvent) + subCtx, subCancelFn := context.WithCancel(svc.Ctx) + cancelFnMap[subscription.Uuid] = subCancelFn + go svc.startSubscription(subCtx, &subscription, nil, handleEvent) } + svc.subCancelFnMap = cancelFnMap return svc, nil } @@ -571,7 +571,11 @@ func (svc *Service) NIP47NotificationHandler(c echo.Context) error { }) } - go svc.startSubscription(svc.Ctx, &subscription, nil, svc.handleSubscribedEvent) + subCtx, subCancelFn := context.WithCancel(svc.Ctx) + svc.subscriptionsMutex.Lock() + svc.subCancelFnMap[subscription.Uuid] = subCancelFn + svc.subscriptionsMutex.Unlock() + go svc.startSubscription(subCtx, &subscription, nil, svc.handleSubscribedEvent) return c.JSON(http.StatusOK, SubscriptionResponse{ SubscriptionId: subscription.Uuid, @@ -630,7 +634,11 @@ func (svc *Service) SubscriptionHandler(c echo.Context) error { }) } - go svc.startSubscription(svc.Ctx, &subscription, nil, svc.handleSubscribedEvent) + subCtx, subCancelFn := context.WithCancel(svc.Ctx) + svc.subscriptionsMutex.Lock() + svc.subCancelFnMap[subscription.Uuid] = subCancelFn + svc.subscriptionsMutex.Unlock() + go svc.startSubscription(subCtx, &subscription, nil, svc.handleSubscribedEvent) return c.JSON(http.StatusOK, SubscriptionResponse{ SubscriptionId: subscription.Uuid, @@ -688,14 +696,17 @@ func (svc *Service) StopSubscriptionHandler(c echo.Context) error { func (svc *Service) stopSubscription(subscription *Subscription) error { svc.subscriptionsMutex.Lock() - sub, exists := svc.subscriptions[subscription.Uuid] + cancelFn, exists := svc.subCancelFnMap[subscription.Uuid] + svc.subscriptionsMutex.Unlock() if exists { - sub.Unsub() - delete(svc.subscriptions, subscription.Uuid) + cancelFn() + } + + if subscription.RelaySubscription != nil { + subscription.RelaySubscription.Unsub() } - svc.subscriptionsMutex.Unlock() - if (!exists && !subscription.Open) { + if (!subscription.Open) { return errors.New(SUBSCRIPTION_ALREADY_CLOSED) } @@ -738,14 +749,12 @@ func (svc *Service) startSubscription(ctx context.Context, subscription *Subscri continue } - sub, err := relay.Subscribe(ctx, []nostr.Filter{*filter}) + relaySubscription, err := relay.Subscribe(ctx, []nostr.Filter{*filter}) if err != nil { continue } - svc.subscriptionsMutex.Lock() - svc.subscriptions[subscription.Uuid] = sub - svc.subscriptionsMutex.Unlock() + subscription.RelaySubscription = relaySubscription svc.Logger.WithFields(logrus.Fields{ "request_event_id": requestEventId, @@ -784,10 +793,8 @@ func (svc *Service) startSubscription(ctx context.Context, subscription *Subscri func (svc *Service) publishRequestEvent(ctx context.Context, subscription *Subscription) { walletPubkey, clientPubkey := getPubkeys(subscription) - svc.subscriptionsMutex.Lock() - sub := svc.subscriptions[subscription.Uuid] - svc.subscriptionsMutex.Unlock() - err := sub.Relay.Publish(ctx, *subscription.RequestEvent.SignedEvent) + relaySubscription := subscription.RelaySubscription + err := relaySubscription.Relay.Publish(ctx, *subscription.RequestEvent.SignedEvent) if err != nil { // TODO: notify user about publish failure svc.Logger.WithError(err).WithFields(logrus.Fields{ @@ -798,7 +805,7 @@ func (svc *Service) publishRequestEvent(ctx context.Context, subscription *Subsc "client_pubkey": clientPubkey, }).Error("Failed to publish to relay") subscription.RequestEvent.State = REQUEST_EVENT_PUBLISH_FAILED - sub.Unsub() + relaySubscription.Unsub() } else { svc.Logger.WithFields(logrus.Fields{ "subscription_id": subscription.Uuid, @@ -855,24 +862,23 @@ func (svc *Service) handleSubscribedEvent(event *nostr.Event, subscription *Subs } func (svc *Service) processEvents(ctx context.Context, subscription *Subscription, onReceiveEOS OnReceiveEOSFunc, handleEvent HandleEventFunc) error { - svc.subscriptionsMutex.Lock() - sub := svc.subscriptions[subscription.Uuid] - svc.subscriptionsMutex.Unlock() + relaySubscription := subscription.RelaySubscription go func(){ - // block till EOS is received - <-sub.EndOfStoredEvents - svc.Logger.WithFields(logrus.Fields{ - "subscription_id": subscription.Uuid, - "relay_url": subscription.RelayUrl, - }).Debug("Received EOS") + // block till EOS is received for nip 47 handlers + // only if request event is not yet published + if (onReceiveEOS != nil && subscription.RequestEvent.State != REQUEST_EVENT_PUBLISH_CONFIRMED) { + <-relaySubscription.EndOfStoredEvents + svc.Logger.WithFields(logrus.Fields{ + "subscription_id": subscription.Uuid, + "relay_url": subscription.RelayUrl, + }).Debug("Received EOS") - if (onReceiveEOS != nil) { onReceiveEOS(ctx, subscription) } // loop through incoming events - for event := range sub.Events { + for event := range relaySubscription.Events { go handleEvent(event, subscription) } @@ -883,11 +889,11 @@ func (svc *Service) processEvents(ctx context.Context, subscription *Subscriptio }() select { - case <-sub.Relay.Context().Done(): - return sub.Relay.ConnectionError + case <-relaySubscription.Relay.Context().Done(): + return relaySubscription.Relay.ConnectionError case <-ctx.Done(): return nil - case <-sub.Context.Done(): + case <-relaySubscription.Context.Done(): return nil } } diff --git a/internal/nostr/push.go b/internal/nostr/push.go index 1439f90..d473528 100644 --- a/internal/nostr/push.go +++ b/internal/nostr/push.go @@ -1,6 +1,7 @@ package nostr import ( + "context" "net/http" "time" @@ -107,7 +108,11 @@ func (svc *Service) NIP47PushNotificationHandler(c echo.Context) error { }) } - go svc.startSubscription(svc.Ctx, &subscription, nil, svc.handleSubscribedEventForPushNotification) + subCtx, subCancelFn := context.WithCancel(svc.Ctx) + svc.subscriptionsMutex.Lock() + svc.subCancelFnMap[subscription.Uuid] = subCancelFn + svc.subscriptionsMutex.Unlock() + go svc.startSubscription(subCtx, &subscription, nil, svc.handleSubscribedEventForPushNotification) return c.JSON(http.StatusOK, PushSubscriptionResponse{ SubscriptionId: subscription.Uuid,