package service import ( "context" "errors" "fmt" "net/http" "path" "strconv" "time" "github.com/getAlby/hub/db" "github.com/getAlby/hub/nip47/models" "github.com/getAlby/hub/swaps" "github.com/getAlby/hub/version" "github.com/getAlby/go-nostr" "github.com/getAlby/go-nostr/nip19" "github.com/sirupsen/logrus" "github.com/getAlby/hub/config" "github.com/getAlby/hub/events" "github.com/getAlby/hub/lnclient" "github.com/getAlby/hub/lnclient/bark" "github.com/getAlby/hub/lnclient/cashu" "github.com/getAlby/hub/lnclient/cln" "github.com/getAlby/hub/lnclient/ldk" "github.com/getAlby/hub/lnclient/lnd" "github.com/getAlby/hub/lnclient/phoenixd" "github.com/getAlby/hub/logger" ) func (svc *service) startNostr(ctx context.Context) error { relayUrls := svc.cfg.GetRelayUrls() if len(relayUrls) == 0 { return errors.New("No relay URLs found") } npub, err := nip19.EncodePublicKey(svc.keys.GetNostrPublicKey()) if err != nil { logger.Logger.WithError(err).Error("Error converting nostr privkey to pubkey") return err } logger.Logger.WithFields(logrus.Fields{ "npub": npub, "hex": svc.keys.GetNostrPublicKey(), "version": version.Tag, "relay_urls": relayUrls, }).Info("Starting Alby Hub") // To debug go-nostr, run with -tags "debug dev" (dev tag so LND build doesn't break with debug tag set) // go run -tags "debug dev" -ldflags="-X 'github.com/getAlby/hub/version.Tag=v1.20.0'" cmd/http/main.go if logger.Logger.GetLevel() >= logrus.DebugLevel { nostr.InfoLogger.SetOutput(logger.Logger.Out) nostr.DebugLogger.SetOutput(logger.Logger.Out) } // Start infinite loop which will be only broken by canceling ctx (SIGINT) pool := nostr.NewSimplePool(ctx, nostr.WithRelayOptions( nostr.WithNoticeHandler(svc.noticeHandler), nostr.WithRequestHeader(http.Header{ "User-Agent": {"AlbyHub/" + version.Tag}, }), )) // initially try connect to relays (if hub has no apps, pool won't connect to relays by default) for _, relayUrl := range svc.cfg.GetRelayUrls() { _, err := pool.EnsureRelay(relayUrl) if err != nil { logger.Logger.WithError(err).WithField("relay_url", relayUrl).Error("failed to initially connect to relay") } } go func() { for { select { case <-ctx.Done(): return default: svc.relayStatuses = nil for _, relayUrl := range svc.cfg.GetRelayUrls() { relay, ok := pool.Relays.Load(relayUrl) svc.relayStatuses = append(svc.relayStatuses, RelayStatus{ Url: relayUrl, Online: ok && relay != nil && relay.IsConnected(), }) } time.Sleep(10 * time.Second) } } }() svc.nip47Service.StartNotifier(ctx, pool) svc.nip47Service.StartNip47InfoPublisher(ctx, pool, svc.GetLNClient()) // register a subscriber for events of "nwc_app_created" which handles creation of nostr subscription for new app createAppEventListener := &createAppConsumer{svc: svc, pool: pool} svc.eventPublisher.RegisterSubscriber(createAppEventListener) // register a subscriber for events of "nwc_app_updated" which handles re-publishing of nip47 event info updateAppEventListener := &updateAppConsumer{svc: svc} svc.eventPublisher.RegisterSubscriber(updateAppEventListener) // start each app wallet subscription which have a child derived wallet key svc.startAllExistingAppsWalletSubscriptions(ctx, pool) // check if there are still legacy apps in DB var legacyAppCount int64 result := svc.db.Model(&db.App{}).Where("wallet_pubkey IS NULL").Count(&legacyAppCount) if result.Error != nil { logger.Logger.WithError(result.Error).Error("Failed to count Legacy Apps") } if legacyAppCount > 0 { go func() { logger.Logger.WithField("legacy_app_count", legacyAppCount).Info("Starting legacy app subscription") // legacy single wallet subscription - only subscribe once for all legacy apps // to ensure we do not get duplicate events svc.startAppWalletSubscription(ctx, pool, svc.keys.GetNostrPublicKey()) }() } go func() { <-ctx.Done() logger.Logger.Info("Main context cancelled, exiting...") pool.Close("exiting") logger.Logger.Info("Relay subroutine ended") svc.eventPublisher.RemoveSubscriber(createAppEventListener) svc.eventPublisher.RemoveSubscriber(updateAppEventListener) }() return nil } // In case the relay somehow loses events or the hub updates with // new capabilities, we re-publish info events for all apps on startup // to ensure that they are retrievable for all connections func (svc *service) publishAllAppInfoEvents() { func() { var legacyAppCount int64 result := svc.db.Model(&db.App{}).Where("wallet_pubkey IS NULL").Count(&legacyAppCount) if result.Error != nil { logger.Logger.WithError(result.Error).Error("Failed to fetch App records with empty WalletPubkey") return } if legacyAppCount > 0 { logger.Logger.WithField("legacy_app_count", legacyAppCount).Debug("Enqueuing publish of legacy info event") for _, relayUrl := range svc.cfg.GetRelayUrls() { svc.nip47Service.EnqueueNip47InfoPublishRequest(0 /* unused */, svc.keys.GetNostrPublicKey(), svc.keys.GetNostrSecretKey(), relayUrl) } } }() var apps []db.App result := svc.db.Where("wallet_pubkey IS NOT NULL").Find(&apps) if result.Error != nil { logger.Logger.WithError(result.Error).Error("Failed to fetch App records with non-empty WalletPubkey") return } for _, app := range apps { func(app db.App) { // queue info event publish request for all existing apps walletPrivKey, err := svc.keys.GetAppWalletKey(app.ID) if err != nil { logger.Logger.WithError(err).WithFields(logrus.Fields{ "app_id": app.ID}).Error("Could not get app wallet key") return } logger.Logger.WithField("app_id", app.ID).Debug("Enqueuing publish of app info event") for _, relayUrl := range svc.cfg.GetRelayUrls() { svc.nip47Service.EnqueueNip47InfoPublishRequest(app.ID, *app.WalletPubkey, walletPrivKey, relayUrl) } }(app) } } func (svc *service) startAllExistingAppsWalletSubscriptions(ctx context.Context, pool *nostr.SimplePool) { var apps []db.App result := svc.db.Where("wallet_pubkey IS NOT NULL").Find(&apps) if result.Error != nil { logger.Logger.WithError(result.Error).Error("Failed to fetch App records with non-empty WalletPubkey") return } for _, app := range apps { go func(app db.App) { svc.startAppWalletSubscription(ctx, pool, *app.WalletPubkey) }(app) } } func (svc *service) startAppWalletSubscription(ctx context.Context, pool *nostr.SimplePool, appWalletPubKey string) error { logger.Logger.Info("Subscribing to events for wallet ", appWalletPubKey) filter := nostr.Filter{ Tags: nostr.TagMap{"p": []string{appWalletPubKey}}, Kinds: []int{models.REQUEST_KIND}, } for { subCtx, cancelSubscription := context.WithCancel(ctx) eventsChannel := pool.SubscribeMany(subCtx, svc.cfg.GetRelayUrls(), filter) // register a subscriber for "nwc_app_deleted" events, which handles // cancelling the nostr subscription and nip47 info event deletion deleteAppSubscriber := deleteAppConsumer{ cancelSubscription: cancelSubscription, walletPubkey: appWalletPubKey, svc: svc, pool: pool, } svc.eventPublisher.RegisterSubscriber(&deleteAppSubscriber) err := svc.watchSubscription(subCtx, pool, eventsChannel) svc.eventPublisher.RemoveSubscriber(&deleteAppSubscriber) if err != nil { logger.Logger.WithError(err).Error("got an error from the relay while listening to subscription, resubscribing") time.Sleep(3 * time.Second) continue } break } return nil } func (svc *service) watchSubscription(ctx context.Context, pool *nostr.SimplePool, eventsChannel chan nostr.RelayEvent) error { eventsChannelClosed := make(chan struct{}) go func() { // loop through incoming events for event := range eventsChannel { select { case <-ctx.Done(): return default: go svc.nip47Service.HandleEvent(ctx, pool, event.Event, svc.GetLNClient()) } } logger.Logger.Debug("Relay subscription events channel ended") eventsChannelClosed <- struct{}{} }() select { case <-ctx.Done(): logger.Logger.Info("Exiting subscription due to context exit...") return nil case <-eventsChannelClosed: // in go-nostr pool, currently if the relay sends a close that is not "auth-required:" // this will trigger closing the subscription channel. We return an error to trigger a resubscribe. logger.Logger.Info("Subscription was exited abnormally") return errors.New("subscription exited abnormally") } } func (svc *service) StartApp(encryptionKey string) error { defer func() { svc.startupState = "" }() svc.startupState = "Initializing" albyIdentifier, err := svc.albyOAuthSvc.GetUserIdentifier() if err != nil { return err } if albyIdentifier != "" && !svc.albyOAuthSvc.IsConnected(svc.ctx) { return errors.New("alby account is not authenticated") } if svc.lnClient != nil { return errors.New("app already started") } unlockPasswordCheckSet, err := svc.cfg.IsUnlockPasswordCheckSet() if err != nil { logger.Logger.WithError(err).Error("Failed to check unlock password check") return fmt.Errorf("check unlock password check: %w", err) } if !unlockPasswordCheckSet { logger.Logger.Error("Unlock password check is missing from the database") return errors.New("your wallet data is incomplete and cannot be unlocked. Please restore from a backup") } if !svc.cfg.CheckUnlockPassword(encryptionKey) { logger.Logger.Errorf("Invalid password") return errors.New("invalid password") } err = svc.cfg.LoadJWTSecret(encryptionKey) if err != nil { logger.Logger.WithError(err).Error("Failed to load JWT secret") return err } err = svc.keys.Init(svc.cfg, encryptionKey) if err != nil { logger.Logger.WithError(err).Error("Failed to init nostr keys") return err } ctx, cancelFn := context.WithCancel(svc.ctx) svc.startupState = "Launching Node" err = svc.launchLNBackend(ctx, encryptionKey) if err != nil { logger.Logger.Errorf("Failed to launch LN backend: %v", err) svc.eventPublisher.Publish(&events.Event{ Event: "nwc_node_start_failed", }) cancelFn() return err } svc.swapsService = swaps.NewSwapsService(ctx, svc.db, svc.cfg, svc.keys, svc.eventPublisher, svc.GetLNClient(), svc.transactionsService, encryptionKey) svc.publishAllAppInfoEvents() svc.startupState = "Connecting To Relay" err = svc.startNostr(ctx) if err != nil { cancelFn() return err } svc.appCancelFn = cancelFn return nil } func (svc *service) launchLNBackend(ctx context.Context, encryptionKey string) error { if svc.lnClient != nil { logger.Logger.Error("LNClient already started") return errors.New("LNClient already started") } svc.wg.Add(1) go func() { // ensure the LNClient is stopped properly before exiting <-ctx.Done() svc.stopLNClient() }() lnBackend, _ := svc.cfg.Get("LNBackendType", "") if lnBackend == "" { return errors.New("no LNBackendType specified") } logger.Logger.Infof("Launching LN Backend: %s", lnBackend) var lnClient lnclient.LNClient var err error vssEnabled := false switch lnBackend { case config.LNDBackendType: LNDAddress, _ := svc.cfg.Get("LNDAddress", encryptionKey) LNDCertHex, _ := svc.cfg.Get("LNDCertHex", encryptionKey) LNDMacaroonHex, _ := svc.cfg.Get("LNDMacaroonHex", encryptionKey) lnClient, err = lnd.NewLNDService(ctx, svc.eventPublisher, LNDAddress, LNDCertHex, LNDMacaroonHex) case config.LDKBackendType: mnemonic, _ := svc.cfg.Get("Mnemonic", encryptionKey) ldkWorkdir := path.Join(svc.cfg.GetEnv().Workdir, "ldk") var vssToken string vssToken, err = svc.requestVssToken(ctx) if err != nil { logger.Logger.WithError(err).Error("Failed to request VSS token") return err } vssEnabled = vssToken != "" svc.startupState = "Launching Node" setStartupState := func(startupState string) { svc.startupState = startupState } channelPeerSuggestions, suggestionsErr := svc.albySvc.GetChannelPeerSuggestions(ctx) if suggestionsErr != nil { logger.Logger.WithError(suggestionsErr).Warn("Failed to fetch channel peer suggestions for LSPS2 liquidity source") } lnClient, err = ldk.NewLDKService(ctx, svc.cfg, svc.eventPublisher, mnemonic, ldkWorkdir, vssToken, setStartupState, channelPeerSuggestions) case config.PhoenixBackendType: PhoenixdAddress, _ := svc.cfg.Get("PhoenixdAddress", encryptionKey) PhoenixdAuthorization, _ := svc.cfg.Get("PhoenixdAuthorization", encryptionKey) lnClient, err = phoenixd.NewPhoenixService(ctx, PhoenixdAddress, PhoenixdAuthorization) case config.CashuBackendType: mnemonic, _ := svc.cfg.Get("Mnemonic", encryptionKey) cashuMintUrl, _ := svc.cfg.Get("CashuMintUrl", encryptionKey) cashuWorkdir := path.Join(svc.cfg.GetEnv().Workdir, "cashu") lnClient, err = cashu.NewCashuService(svc.cfg, cashuWorkdir, mnemonic, cashuMintUrl) case config.BarkBackendType: mnemonic, _ := svc.cfg.Get("Mnemonic", encryptionKey) env := svc.cfg.GetEnv() barkWorkdir := path.Join(env.Workdir, "bark") lnClient, err = bark.NewBarkService(ctx, svc.eventPublisher, barkWorkdir, mnemonic, bark.Config{ Network: svc.cfg.GetNetwork(), ServerAddress: env.BarkServer, EsploraAddress: env.BarkEsploraServer, ServerAccessToken: env.BarkServerAccessToken, LogLevel: env.BarkLogLevel, LogToFile: env.LogToFile, }) case config.CLNBackendType: CLNAddress, _ := svc.cfg.Get("CLNAddress", encryptionKey) CLNLightningDir, _ := svc.cfg.Get("CLNLightningDir", encryptionKey) CLNAddressHold, _ := svc.cfg.Get("CLNAddressHold", encryptionKey) lnClient, err = cln.NewCLNService(ctx, svc.eventPublisher, CLNAddress, CLNLightningDir, CLNAddressHold) default: logger.Logger.WithField("backend_type", lnBackend).Error("Unsupported LNBackendType") return fmt.Errorf("unsupported backend type: %s", lnBackend) } if err != nil { logger.Logger.WithError(err).Error("Failed to launch LN backend") return err } // TODO: call a method on the LNClient here to check the LNClient is actually connectable, // (e.g. lnClient.CheckConnection()) Rather than it being a side-effect // in the LNClient init function svc.lnClient = lnClient info, err := lnClient.GetInfo(ctx) if err != nil { logger.Logger.WithError(err).Error("Failed to fetch node info") } if info != nil { svc.eventPublisher.SetGlobalProperty("node_id", info.Pubkey) svc.eventPublisher.SetGlobalProperty("network", info.Network) } // Mark that the node has successfully started // This will ensure the user cannot go through the setup again err = svc.cfg.SetUpdate("NodeLastStartTime", strconv.FormatInt(time.Now().Unix(), 10), "") if err != nil { logger.Logger.WithError(err).Error("Failed to set last node start time") } svc.eventPublisher.Publish(&events.Event{ Event: "nwc_node_started", Properties: map[string]interface{}{ "node_type": lnBackend, "vss_enabled": vssEnabled, }, }) return nil } func (svc *service) requestVssToken(ctx context.Context) (string, error) { nodeLastStartTime, _ := svc.cfg.Get("NodeLastStartTime", "") // for brand new nodes, consider enabling VSS if nodeLastStartTime == "" && svc.cfg.GetEnv().LDKVssUrl != "" { svc.startupState = "Checking Subscription" albyUserIdentifier, err := svc.albyOAuthSvc.GetUserIdentifier() if err != nil { logger.Logger.WithError(err).Error("Failed to fetch alby user identifier") return "", err } if albyUserIdentifier != "" { me, err := svc.albyOAuthSvc.GetMe(ctx) if err != nil { logger.Logger.WithError(err).Error("Failed to fetch alby user") return "", err } // only activate VSS for Alby paid subscribers if me.Subscription.PlanCode != "" { svc.cfg.SetUpdate("LdkVssEnabled", "true", "") } } } vssToken := "" vssEnabled, _ := svc.cfg.Get("LdkVssEnabled", "") if vssEnabled == "true" { svc.startupState = "Fetching VSS token" vssNodeIdentifier, err := ldk.GetVssNodeIdentifier(svc.keys) if err != nil { logger.Logger.WithError(err).Error("Failed to get VSS node identifier") return "", err } vssToken, err = svc.albyOAuthSvc.GetVssAuthToken(ctx, vssNodeIdentifier) if err != nil { logger.Logger.WithError(err).Error("Failed to fetch VSS JWT token") existingVssToken, _ := svc.cfg.Get("VssToken", "") if existingVssToken != "" { logger.Logger.Warn("Using stored VSS JWT token") return existingVssToken, nil } return "", err } err = svc.cfg.SetUpdate("VssToken", vssToken, "") if err != nil { logger.Logger.WithError(err).Error("Failed to save VSS JWT token to user config") } } return vssToken, nil }