Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

chore: change status to publishstatus #88

Merged
merged 3 commits into from
Jul 15, 2024
Merged

chore: change status to publishstatus #88

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ffb0f30
chore: change status to publishstatus
im-adithya Jul 15, 2024
0f23f96
chore: use snake_case for log fields
im-adithya Jul 15, 2024
634f920
chore: add wallet pubkey to logs
im-adithya Jul 15, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
148 changes: 76 additions & 72 deletions internal/nostr/nostr.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -175,8 +175,8 @@ func (svc *Service) InfoHandler(c echo.Context) error {
}

svc.Logger.WithFields(logrus.Fields{
"relayUrl": requestData.RelayUrl,
"walletPubkey": requestData.WalletPubkey,
"relay_url": requestData.RelayUrl,
"wallet_pubkey": requestData.WalletPubkey,
}).Info("Subscribing to info event")

filter := nostr.Filter{
Expand All @@ -195,18 +195,18 @@ func (svc *Service) InfoHandler(c echo.Context) error {
select {
case <-ctx.Done():
svc.Logger.WithFields(logrus.Fields{
"relayUrl": requestData.RelayUrl,
"walletPubkey": requestData.WalletPubkey,
"relay_url": requestData.RelayUrl,
"wallet_pubkey": requestData.WalletPubkey,
}).Info("Exiting info subscription without receiving")
return c.JSON(http.StatusRequestTimeout, ErrorResponse{
Message: "Request canceled or timed out",
Error: ctx.Err().Error(),
})
case event := <-sub.Events:
svc.Logger.WithFields(logrus.Fields{
"relayUrl": requestData.RelayUrl,
"walletPubkey": requestData.WalletPubkey,
"eventId": event.ID,
"relay_url": requestData.RelayUrl,
"wallet_pubkey": requestData.WalletPubkey,
"event_id": event.ID,
}).Info("Received info event")
sub.Unsub()
return c.JSON(http.StatusOK, InfoResponse{
Expand Down Expand Up @@ -247,15 +247,15 @@ func (svc *Service) PublishHandler(c echo.Context) error {
}

svc.Logger.WithFields(logrus.Fields{
"eventId": requestData.SignedEvent.ID,
"relayUrl": requestData.RelayUrl,
"event_id": requestData.SignedEvent.ID,
"relay_url": requestData.RelayUrl,
}).Info("Publishing event")

err = relay.Publish(ctx, *requestData.SignedEvent)
if err != nil {
svc.Logger.WithError(err).WithFields(logrus.Fields{
"eventId": requestData.SignedEvent.ID,
"relayUrl": requestData.RelayUrl,
"event_id": requestData.SignedEvent.ID,
"relay_url": requestData.RelayUrl,
}).Error("Failed to publish event")

return c.JSON(http.StatusInternalServerError, ErrorResponse{
Expand All @@ -265,8 +265,8 @@ func (svc *Service) PublishHandler(c echo.Context) error {
}

svc.Logger.WithFields(logrus.Fields{
"eventId": requestData.SignedEvent.ID,
"relayUrl": requestData.RelayUrl,
"event_id": requestData.SignedEvent.ID,
"relay_url": requestData.RelayUrl,
}).Info("Published event")

return c.JSON(http.StatusOK, PublishResponse{
Expand Down Expand Up @@ -300,10 +300,9 @@ func (svc *Service) NIP47Handler(c echo.Context) error {
}

svc.Logger.WithFields(logrus.Fields{
"requestEventId": requestData.SignedEvent.ID,
"connectionPubkey": requestData.SignedEvent.PubKey,
"walletPubkey": requestData.WalletPubkey,
"relayUrl": requestData.RelayUrl,
"request_event_id": requestData.SignedEvent.ID,
"wallet_pubkey": requestData.WalletPubkey,
"relay_url": requestData.RelayUrl,
}).Info("Processing request event")

if svc.db.Where("nostr_id = ?", requestData.SignedEvent.ID).Find(&RequestEvent{}).RowsAffected != 0 {
Expand Down Expand Up @@ -337,10 +336,9 @@ func (svc *Service) NIP47Handler(c echo.Context) error {
select {
case <-ctx.Done():
svc.Logger.WithFields(logrus.Fields{
"requestEventId": requestData.SignedEvent.ID,
"connectionPubkey": requestData.SignedEvent.PubKey,
"walletPubkey": requestData.WalletPubkey,
"relayUrl": requestData.RelayUrl,
"request_event_id": requestData.SignedEvent.ID,
"wallet_pubkey": requestData.WalletPubkey,
"relay_url": requestData.RelayUrl,
}).Info("Stopped subscription without receiving event")
if ctx.Err() == context.DeadlineExceeded {
return c.JSON(http.StatusGatewayTimeout, ErrorResponse{
Expand All @@ -354,11 +352,10 @@ func (svc *Service) NIP47Handler(c echo.Context) error {
})
case event := <-subscription.EventChan:
svc.Logger.WithFields(logrus.Fields{
"requestEventId": requestData.SignedEvent.ID,
"connectionPubkey": requestData.SignedEvent.PubKey,
"responseEventId": event.ID,
"walletPubkey": requestData.WalletPubkey,
"relayUrl": requestData.RelayUrl,
"request_event_id": requestData.SignedEvent.ID,
"response_event_id": event.ID,
"wallet_pubkey": requestData.WalletPubkey,
"relay_url": requestData.RelayUrl,
}).Info("Received response event")
return c.JSON(http.StatusOK, NIP47Response{
Event: event,
Expand Down Expand Up @@ -398,11 +395,10 @@ func (svc *Service) NIP47WebhookHandler(c echo.Context) error {
}

svc.Logger.WithFields(logrus.Fields{
"requestEventId": requestData.SignedEvent.ID,
"connectionPubkey": requestData.SignedEvent.PubKey,
"walletPubkey": requestData.WalletPubkey,
"relayUrl": requestData.RelayUrl,
"webhookUrl": requestData.WebhookUrl,
"request_event_id": requestData.SignedEvent.ID,
"wallet_pubkey": requestData.WalletPubkey,
"relay_url": requestData.RelayUrl,
"webhook_url": requestData.WebhookUrl,
}).Info("Processing request event")

if svc.db.Where("nostr_id = ?", requestData.SignedEvent.ID).First(&RequestEvent{}).RowsAffected != 0 {
Expand Down Expand Up @@ -483,10 +479,9 @@ func (svc *Service) NIP47NotificationHandler(c echo.Context) error {
}

svc.Logger.WithFields(logrus.Fields{
"connectionPubkey": requestData.ConnPubkey,
"walletPubkey": requestData.WalletPubkey,
"relayUrl": requestData.RelayUrl,
"webhookUrl": requestData.WebhookUrl,
"wallet_pubkey": requestData.WalletPubkey,
"relay_url": requestData.RelayUrl,
"webhook_url": requestData.WebhookUrl,
}).Info("Subscribing to notifications")

subscription := Subscription{
Expand Down Expand Up @@ -598,13 +593,13 @@ func (svc *Service) StopSubscriptionHandler(c echo.Context) error {
}

svc.Logger.WithFields(logrus.Fields{
"subscriptionId": subscription.ID,
"subscription_id": subscription.ID,
}).Info("Stopping subscription")

err := svc.stopSubscription(&subscription)
if err != nil {
svc.Logger.WithFields(logrus.Fields{
"subscriptionId": subscription.ID,
"subscription_id": subscription.ID,
}).Info("Subscription is stopped already")

return c.JSON(http.StatusAlreadyReported, StopSubscriptionResponse{
Expand All @@ -618,7 +613,7 @@ func (svc *Service) StopSubscriptionHandler(c echo.Context) error {
// delete svix app

svc.Logger.WithFields(logrus.Fields{
"subscriptionId": subscription.ID,
"subscription_id": subscription.ID,
}).Info("Stopped subscription")

return c.JSON(http.StatusOK, StopSubscriptionResponse{
Expand All @@ -645,7 +640,8 @@ func (svc *Service) stopSubscription(subscription *Subscription) error {

func (svc *Service) startSubscription(ctx context.Context, subscription *Subscription, onReceiveEOS OnReceiveEOSFunc, handleEvent HandleEventFunc) {
svc.Logger.WithFields(logrus.Fields{
"subscriptionId": subscription.ID,
"subscription_id": subscription.ID,
"wallet_pubkey": (*subscription.Authors)[0],
}).Info("Starting subscription")

filter := svc.subscriptionToFilter(subscription)
Expand All @@ -665,8 +661,9 @@ func (svc *Service) startSubscription(ctx context.Context, subscription *Subscri
if err != nil {
// TODO: notify user about relay failure
svc.Logger.WithError(err).WithFields(logrus.Fields{
"subscriptionId": subscription.ID,
"relayUrl": subscription.RelayUrl,
"subscription_id": subscription.ID,
"relay_url": subscription.RelayUrl,
"wallet_pubkey": (*subscription.Authors)[0],
}).Error("Failed get relay connection, retrying in 5s...")
time.Sleep(5 * time.Second) // sleep for 5 seconds
continue
Expand All @@ -676,8 +673,9 @@ func (svc *Service) startSubscription(ctx context.Context, subscription *Subscri
if err != nil {
// TODO: notify user about subscription failure
svc.Logger.WithError(err).WithFields(logrus.Fields{
"subscriptionId": subscription.ID,
"relayUrl": subscription.RelayUrl,
"subscription_id": subscription.ID,
"relay_url": subscription.RelayUrl,
"wallet_pubkey": (*subscription.Authors)[0],
}).Error("Failed to subscribe to relay, retrying in 5s...")
time.Sleep(5 * time.Second) // sleep for 5 seconds
continue
Expand All @@ -688,16 +686,18 @@ func (svc *Service) startSubscription(ctx context.Context, subscription *Subscri
svc.subscriptionsMutex.Unlock()

svc.Logger.WithFields(logrus.Fields{
"subscriptionId": subscription.ID,
"subscription_id": subscription.ID,
"wallet_pubkey": (*subscription.Authors)[0],
}).Info("Started subscription")

err = svc.processEvents(ctx, subscription, onReceiveEOS, handleEvent)

if err != nil {
// TODO: notify user about subscription failure
svc.Logger.WithError(err).WithFields(logrus.Fields{
"subscriptionId": subscription.ID,
"relayUrl": subscription.RelayUrl,
"subscription_id": subscription.ID,
"relay_url": subscription.RelayUrl,
"wallet_pubkey": (*subscription.Authors)[0],
}).Error("Subscription stopped due to relay error, reconnecting in 5s...")
time.Sleep(5 * time.Second) // sleep for 5 seconds
continue
Expand All @@ -716,8 +716,9 @@ func (svc *Service) startSubscription(ctx context.Context, subscription *Subscri
svc.stopSubscription(subscription)
}
svc.Logger.WithFields(logrus.Fields{
"subscriptionId": subscription.ID,
"relayUrl": subscription.RelayUrl,
"subscription_id": subscription.ID,
"relay_url": subscription.RelayUrl,
"wallet_pubkey": (*subscription.Authors)[0],
}).Info("Stopping subscription")
break
}
Expand All @@ -732,26 +733,27 @@ func (svc *Service) publishEvent(ctx context.Context, subscription *Subscription
if err != nil {
// TODO: notify user about publish failure
svc.Logger.WithError(err).WithFields(logrus.Fields{
"subscriptionId": subscription.ID,
"relayUrl": subscription.RelayUrl,
"subscription_id": subscription.ID,
"relay_url": subscription.RelayUrl,
"wallet_pubkey": (*subscription.Authors)[0],
}).Error("Failed to publish to relay")
sub.Unsub()
} else {
svc.Logger.WithFields(logrus.Fields{
"status": REQUEST_EVENT_PUBLISH_CONFIRMED,
"eventId": subscription.RequestEvent.ID,
"connectionPubkey": subscription.RequestEvent.PubKey,
"publish_status": REQUEST_EVENT_PUBLISH_CONFIRMED,
"event_id": subscription.RequestEvent.ID,
"wallet_pubkey": (*subscription.Authors)[0],
}).Info("Published request event successfully")
subscription.RequestEventDB.State = REQUEST_EVENT_PUBLISH_CONFIRMED
}
}

func (svc *Service) handleResponseEvent(event *nostr.Event, subscription *Subscription) {
svc.Logger.WithFields(logrus.Fields{
"eventId": event.ID,
"eventKind": event.Kind,
"requestEventId": subscription.RequestEvent.ID,
"connectionPubkey": subscription.RequestEvent.PubKey,
"event_id": event.ID,
"event_kind": event.Kind,
"request_event_id": subscription.RequestEvent.ID,
"wallet_pubkey": (*subscription.Authors)[0],
}).Info("Received response event")
responseEvent := ResponseEvent{
NostrId: event.ID,
Expand All @@ -772,9 +774,10 @@ func (svc *Service) handleResponseEvent(event *nostr.Event, subscription *Subscr

func (svc *Service) handleSubscribedEvent(event *nostr.Event, subscription *Subscription) {
svc.Logger.WithFields(logrus.Fields{
"eventId": event.ID,
"eventKind": event.Kind,
"subscriptionId": subscription.ID,
"event_id": event.ID,
"event_kind": event.Kind,
"subscription_id": subscription.ID,
"wallet_pubkey": (*subscription.Authors)[0],
}).Info("Received event")
responseEvent := ResponseEvent{
NostrId: event.ID,
Expand All @@ -795,7 +798,8 @@ func (svc *Service) processEvents(ctx context.Context, subscription *Subscriptio
// block till EOS is received
<-sub.EndOfStoredEvents
svc.Logger.WithFields(logrus.Fields{
"subscriptionId": subscription.ID,
"subscription_id": subscription.ID,
"wallet_pubkey": (*subscription.Authors)[0],
}).Info("Received EOS")

if (onReceiveEOS != nil) {
Expand All @@ -808,7 +812,7 @@ func (svc *Service) processEvents(ctx context.Context, subscription *Subscriptio
}

svc.Logger.WithFields(logrus.Fields{
"subscriptionId": subscription.ID,
"subscription_id": subscription.ID,
}).Info("Relay subscription events channel ended")
}()

Expand All @@ -825,7 +829,7 @@ func (svc *Service) processEvents(ctx context.Context, subscription *Subscriptio
func (svc *Service) getRelayConnection(ctx context.Context, customRelayURL string) (*nostr.Relay, bool, error) {
if customRelayURL != "" && customRelayURL != svc.Cfg.DefaultRelayURL {
svc.Logger.WithFields(logrus.Fields{
"customRelayURL": customRelayURL,
"custom_relay_url": customRelayURL,
}).Infof("Connecting to custom relay")
relay, err := nostr.RelayConnect(ctx, customRelayURL)
return relay, true, err // true means custom and the relay should be closed
Expand All @@ -851,9 +855,9 @@ func (svc *Service) postEventToWebhook(event *nostr.Event, webhookURL string) {
eventData, err := json.Marshal(event)
if err != nil {
svc.Logger.WithError(err).WithFields(logrus.Fields{
"eventId": event.ID,
"eventKind": event.Kind,
"webhookUrl": webhookURL,
"event_id": event.ID,
"event_kind": event.Kind,
"webhook_url": webhookURL,
}).Error("Failed to marshal event for webhook")
return
}
Expand All @@ -862,16 +866,16 @@ func (svc *Service) postEventToWebhook(event *nostr.Event, webhookURL string) {
_, err = http.Post(webhookURL, "application/json", bytes.NewBuffer(eventData))
if err != nil {
svc.Logger.WithError(err).WithFields(logrus.Fields{
"eventId": event.ID,
"eventKind": event.Kind,
"webhookUrl": webhookURL,
"event_id": event.ID,
"event_kind": event.Kind,
"webhook_url": webhookURL,
}).Error("Failed to post event to webhook")
}

svc.Logger.WithFields(logrus.Fields{
"eventId": event.ID,
"eventKind": event.Kind,
"webhookUrl": webhookURL,
"event_id": event.ID,
"event_kind": event.Kind,
"webhook_url": webhookURL,
}).Infof("Successfully posted event to webhook")
}

Expand Down
Loading