feat: move events to transactions service (WIP)

alby/alby_oauth_service.go

@@ -460,11 +460,6 @@ func (svc *albyOAuthService) LinkAccount(ctx context.Context, lnClient lnclient.
 }
 
 func (svc *albyOAuthService) ConsumeEvent(ctx context.Context, event *events.Event, globalProperties map[string]interface{}) {
-	// run non-blocking
-	go svc.consumeEvent(ctx, event, globalProperties)
-}
-
-func (svc *albyOAuthService) consumeEvent(ctx context.Context, event *events.Event, globalProperties map[string]interface{}) {
 	// TODO: rename this config option to be specific to the alby API
 	if !svc.cfg.GetEnv().LogEvents {
 		logger.Logger.WithField("event", event).Debug("Skipped sending to alby events API")
@@ -478,6 +473,11 @@ func (svc *albyOAuthService) consumeEvent(ctx context.Context, event *events.Eve
 		return
 	}
 
+	if strings.HasPrefix(event.Event, "nwc_lnclient_") {
+		// don't consume internal LNClient events
+		return
+	}
+
 	if event.Event == "nwc_payment_received" {
 		type paymentReceivedEventProperties struct {
 			PaymentHash string `json:"payment_hash"`
@@ -507,24 +507,24 @@ func (svc *albyOAuthService) consumeEvent(ctx context.Context, event *events.Eve
 		}
 	}
 
-	if event.Event == "nwc_payment_failed_async" {
-		paymentFailedAsyncProperties, ok := event.Properties.(*events.PaymentFailedAsyncProperties)
+	if event.Event == "nwc_payment_failed" {
+		transaction, ok := event.Properties.(*db.Transaction)
 		if !ok {
 			logger.Logger.WithField("event", event).Error("Failed to cast event")
 			return
 		}
 
-		type paymentSentEventProperties struct {
+		type paymentFailedEventProperties struct {
 			PaymentHash string `json:"payment_hash"`
 			Reason      string `json:"reason"`
 		}
 
 		// pass a new custom event with less detail
 		event = &events.Event{
 			Event: event.Event,
-			Properties: &paymentSentEventProperties{
-				PaymentHash: paymentFailedAsyncProperties.Transaction.PaymentHash,
-				Reason:      paymentFailedAsyncProperties.Reason,
+			Properties: &paymentFailedEventProperties{
+				PaymentHash: transaction.PaymentHash,
+				Reason:      transaction.FailureReason,
 			},
 		}
 	}

db/migrations/202408191242_transaction_failure_reason.go

@@ -0,0 +1,26 @@
+package migrations
+
+import (
+	_ "embed"
+
+	"github.com/go-gormigrate/gormigrate/v2"
+	"gorm.io/gorm"
+)
+
+// This migration removes old app permissions for request methods (now we use scopes)
+var _202408191242_transaction_failure_reason = &gormigrate.Migration{
+	ID: "202408191242_transaction_failure_reason",
+	Migrate: func(tx *gorm.DB) error {
+
+		if err := tx.Exec(`
+	ALTER TABLE transactions ADD failure_reason string;
+`).Error; err != nil {
+			return err
+		}
+
+		return nil
+	},
+	Rollback: func(tx *gorm.DB) error {
+		return nil
+	},
+}

db/migrations/migrate.go

@@ -20,6 +20,7 @@ func Migrate(gormDB *gorm.DB) error {
 		_202407201604_transactions_indexes,
 		_202407262257_remove_invalid_scopes,
 		_202408061737_add_boostagrams_and_use_json,
+		_202408191242_transaction_error,
 	})
 
 	return m.Migrate()

db/models.go

@@ -82,6 +82,7 @@ type Transaction struct {
 	Metadata        datatypes.JSON
 	SelfPayment     bool
 	Boostagram      datatypes.JSON
+	FailureReason   string
 }
 
 type DBService interface {

events/events.go

@@ -48,9 +48,8 @@ func (ep *eventPublisher) Publish(event *Event) {
 	defer ep.subscriberMtx.Unlock()
 	logger.Logger.WithFields(logrus.Fields{"event": event, "global": ep.globalProperties}).Debug("Publishing event")
 	for _, listener := range ep.listeners {
-		// events are consumed in sequence as some listeners depend on earlier consumers
-		// (e.g. NIP-47 notifier depends on transactions service updating transactions)
-		listener.ConsumeEvent(context.Background(), event, ep.globalProperties)
+		// consume event without blocking thread
+		go listener.ConsumeEvent(context.Background(), event, ep.globalProperties)
 	}
 }
 

events/models.go

@@ -2,8 +2,6 @@ package events
 
 import (
 	"context"
-
-	"github.com/getAlby/hub/lnclient"
 )
 
 type EventSubscriber interface {
@@ -34,8 +32,3 @@ type ChannelBackupInfo struct {
 	FundingTxID   string `json:"funding_tx_id"`
 	FundingTxVout uint32 `json:"funding_tx_vout"`
 }
-
-type PaymentFailedAsyncProperties struct {
-	Transaction *lnclient.Transaction
-	Reason      string
-}

lnclient/ldk/ldk.go

@@ -1309,7 +1309,7 @@ func (ls *LDKService) handleLdkEvent(event *ldk_node.Event) {
 		}
 
 		ls.eventPublisher.Publish(&events.Event{
-			Event:      "nwc_payment_received",
+			Event:      "nwc_lnclient_payment_received",
 			Properties: transaction,
 		})
 	case ldk_node.EventPaymentSuccessful:
@@ -1330,7 +1330,7 @@ func (ls *LDKService) handleLdkEvent(event *ldk_node.Event) {
 		}
 
 		ls.eventPublisher.Publish(&events.Event{
-			Event:      "nwc_payment_sent",
+			Event:      "nwc_lnclient_payment_sent",
 			Properties: transaction,
 		})
 	case ldk_node.EventPaymentFailed:
@@ -1353,8 +1353,8 @@ func (ls *LDKService) handleLdkEvent(event *ldk_node.Event) {
 		reason := ls.getPaymentFailReason(&eventType)
 
 		ls.eventPublisher.Publish(&events.Event{
-			Event: "nwc_payment_failed_async",
-			Properties: &events.PaymentFailedAsyncProperties{
+			Event: "nwc_lnclient_payment_failed",
+			Properties: &lnclient.PaymentFailedEventProperties{
 				Transaction: transaction,
 				Reason:      reason,
 			},

lnclient/lnd/lnd.go

@@ -476,8 +476,8 @@ func NewLNDService(ctx context.Context, eventPublisher events.EventPublisher, ln
 							continue
 						}
 						eventPublisher.Publish(&events.Event{
-							Event: "nwc_payment_failed_async",
-							Properties: &events.PaymentFailedAsyncProperties{
+							Event: "nwc_lnclient_payment_failed",
+							Properties: &lnclient.PaymentFailedEventProperties{
 								Transaction: transaction,
 								Reason:      payment.FailureReason.String(),
 							},
@@ -492,7 +492,7 @@ func NewLNDService(ctx context.Context, eventPublisher events.EventPublisher, ln
 							continue
 						}
 						eventPublisher.Publish(&events.Event{
-							Event:      "nwc_payment_sent",
+							Event:      "nwc_lnclient_payment_sent",
 							Properties: transaction,
 						})
 					default:
@@ -542,7 +542,7 @@ func NewLNDService(ctx context.Context, eventPublisher events.EventPublisher, ln
 					}).Info("Received new invoice")
 
 					eventPublisher.Publish(&events.Event{
-						Event:      "nwc_payment_received",
+						Event:      "nwc_lnclient_payment_received",
 						Properties: lndInvoiceToTransaction(invoice),
 					})
 				}

lnclient/models.go

@@ -171,6 +171,11 @@ type BalancesResponse struct {
 
 type NetworkGraphResponse = interface{}
 
+type PaymentFailedEventProperties struct {
+	Transaction *Transaction
+	Reason      string
+}
+
 // default invoice expiry in seconds (1 day)
 const DEFAULT_INVOICE_EXPIRY = 86400
 

nip47/controllers/pay_invoice_controller.go

@@ -79,14 +79,6 @@ func (controller *nip47Controller) pay(ctx context.Context, bolt11 string, payme
 		return
 	}
 
-	controller.eventPublisher.Publish(&events.Event{
-		Event: "nwc_payment_succeeded",
-		Properties: map[string]interface{}{
-			"bolt11": bolt11,
-			"amount": paymentRequest.MSatoshi / 1000,
-		},
-	})
-
 	publishResponse(&models.Response{
 		ResultType: nip47Request.Method,
 		Result: payResponse{

nip47/controllers/pay_keysend_controller.go

@@ -57,13 +57,7 @@ func (controller *nip47Controller) payKeysend(ctx context.Context, payKeysendPar
 		}, tags)
 		return
 	}
-	controller.eventPublisher.Publish(&events.Event{
-		Event: "nwc_payment_succeeded",
-		Properties: map[string]interface{}{
-			"keysend": true,
-			"amount":  payKeysendParams.Amount / 1000,
-		},
-	})
+
 	publishResponse(&models.Response{
 		ResultType: nip47Request.Method,
 		Result: payResponse{

service/service.go

@@ -104,14 +104,11 @@ func NewService(ctx context.Context) (*service, error) {
 		eventPublisher:      eventPublisher,
 		albyOAuthSvc:        alby.NewAlbyOAuthService(gormDB, cfg, keys, eventPublisher),
 		nip47Service:        nip47.NewNip47Service(gormDB, cfg, keys, eventPublisher),
-		transactionsService: transactions.NewTransactionsService(gormDB),
+		transactionsService: transactions.NewTransactionsService(gormDB, eventPublisher),
 		db:                  gormDB,
 		keys:                keys,
 	}
 
-	// Note: order is important here: transactions service will update transactions
-	// from payment events, which will then be consumed by the NIP-47 service to send notifications
-	// TODO: transactions service should fire its own events
 	eventPublisher.RegisterSubscriber(svc.transactionsService)
 	eventPublisher.RegisterSubscriber(svc.nip47Service)
 	eventPublisher.RegisterSubscriber(svc.albyOAuthSvc)

transactions/notifications_test.go

@@ -33,7 +33,7 @@ func TestNotifications_ReceivedKnownPayment(t *testing.T) {
 	transactionsService := NewTransactionsService(svc.DB)
 
 	transactionsService.ConsumeEvent(ctx, &events.Event{
-		Event:      "nwc_payment_received",
+		Event:      "nwc_lnclient_payment_received",
 		Properties: tests.MockLNClientTransaction,
 	}, map[string]interface{}{})
 
@@ -59,7 +59,7 @@ func TestNotifications_ReceivedUnknownPayment(t *testing.T) {
 	transactionsService := NewTransactionsService(svc.DB)
 
 	transactionsService.ConsumeEvent(ctx, &events.Event{
-		Event:      "nwc_payment_received",
+		Event:      "nwc_lnclient_payment_received",
 		Properties: tests.MockLNClientTransaction,
 	}, map[string]interface{}{})
 
@@ -108,7 +108,7 @@ func TestNotifications_ReceivedKeysend(t *testing.T) {
 	}
 
 	transactionsService.ConsumeEvent(ctx, &events.Event{
-		Event:      "nwc_payment_received",
+		Event:      "nwc_lnclient_payment_received",
 		Properties: transaction,
 	}, map[string]interface{}{})
 
@@ -161,7 +161,7 @@ func TestNotifications_SentKnownPayment(t *testing.T) {
 	transactionsService := NewTransactionsService(svc.DB)
 
 	transactionsService.ConsumeEvent(ctx, &events.Event{
-		Event:      "nwc_payment_sent",
+		Event:      "nwc_lnclient_payment_sent",
 		Properties: tests.MockLNClientTransaction,
 	}, map[string]interface{}{})
 
@@ -192,7 +192,7 @@ func TestNotifications_SentUnknownPayment(t *testing.T) {
 	assert.Equal(t, int64(0), result.RowsAffected)
 
 	transactionsService.ConsumeEvent(ctx, &events.Event{
-		Event:      "nwc_payment_sent",
+		Event:      "nwc_lnclient_payment_sent",
 		Properties: tests.MockLNClientTransaction,
 	}, map[string]interface{}{})
 
@@ -221,8 +221,8 @@ func TestNotifications_FailedKnownPayment(t *testing.T) {
 	transactionsService := NewTransactionsService(svc.DB)
 
 	transactionsService.ConsumeEvent(ctx, &events.Event{
-		Event: "nwc_payment_failed_async",
-		Properties: &events.PaymentFailedAsyncProperties{
+		Event: "nwc_lnclient_payment_failed",
+		Properties: &events.LNClientPaymentFailedAsyncProperties{
 			Transaction: tests.MockLNClientTransaction,
 			Reason:      "Some failure reason",
 		},