start.go raw
1 package service
2
3 import (
4 "context"
5 "errors"
6 "fmt"
7 "net/http"
8 "path"
9 "strconv"
10 "time"
11
12 "github.com/getAlby/hub/db"
13 "github.com/getAlby/hub/nip47/models"
14 "github.com/getAlby/hub/swaps"
15 "github.com/getAlby/hub/version"
16
17 "github.com/getAlby/go-nostr"
18 "github.com/getAlby/go-nostr/nip19"
19 "github.com/sirupsen/logrus"
20
21 "github.com/getAlby/hub/config"
22 "github.com/getAlby/hub/events"
23 "github.com/getAlby/hub/lnclient"
24 "github.com/getAlby/hub/lnclient/bark"
25 "github.com/getAlby/hub/lnclient/cashu"
26 "github.com/getAlby/hub/lnclient/cln"
27 "github.com/getAlby/hub/lnclient/ldk"
28 "github.com/getAlby/hub/lnclient/lnd"
29 "github.com/getAlby/hub/lnclient/phoenixd"
30 "github.com/getAlby/hub/logger"
31 )
32
33 func (svc *service) startNostr(ctx context.Context) error {
34 relayUrls := svc.cfg.GetRelayUrls()
35 if len(relayUrls) == 0 {
36 return errors.New("No relay URLs found")
37 }
38
39 npub, err := nip19.EncodePublicKey(svc.keys.GetNostrPublicKey())
40 if err != nil {
41 logger.Logger.WithError(err).Error("Error converting nostr privkey to pubkey")
42 return err
43 }
44
45 logger.Logger.WithFields(logrus.Fields{
46 "npub": npub,
47 "hex": svc.keys.GetNostrPublicKey(),
48 "version": version.Tag,
49 "relay_urls": relayUrls,
50 }).Info("Starting Alby Hub")
51
52 // To debug go-nostr, run with -tags "debug dev" (dev tag so LND build doesn't break with debug tag set)
53 // go run -tags "debug dev" -ldflags="-X 'github.com/getAlby/hub/version.Tag=v1.20.0'" cmd/http/main.go
54 if logger.Logger.GetLevel() >= logrus.DebugLevel {
55 nostr.InfoLogger.SetOutput(logger.Logger.Out)
56 nostr.DebugLogger.SetOutput(logger.Logger.Out)
57 }
58
59 // Start infinite loop which will be only broken by canceling ctx (SIGINT)
60 pool := nostr.NewSimplePool(ctx, nostr.WithRelayOptions(
61 nostr.WithNoticeHandler(svc.noticeHandler),
62 nostr.WithRequestHeader(http.Header{
63 "User-Agent": {"AlbyHub/" + version.Tag},
64 }),
65 ))
66
67 // initially try connect to relays (if hub has no apps, pool won't connect to relays by default)
68 for _, relayUrl := range svc.cfg.GetRelayUrls() {
69 _, err := pool.EnsureRelay(relayUrl)
70 if err != nil {
71 logger.Logger.WithError(err).WithField("relay_url", relayUrl).Error("failed to initially connect to relay")
72 }
73 }
74 go func() {
75 for {
76 select {
77 case <-ctx.Done():
78 return
79 default:
80 svc.relayStatuses = nil
81 for _, relayUrl := range svc.cfg.GetRelayUrls() {
82 relay, ok := pool.Relays.Load(relayUrl)
83 svc.relayStatuses = append(svc.relayStatuses, RelayStatus{
84 Url: relayUrl,
85 Online: ok && relay != nil && relay.IsConnected(),
86 })
87 }
88 time.Sleep(10 * time.Second)
89 }
90 }
91 }()
92
93 svc.nip47Service.StartNotifier(ctx, pool)
94 svc.nip47Service.StartNip47InfoPublisher(ctx, pool, svc.GetLNClient())
95
96 // register a subscriber for events of "nwc_app_created" which handles creation of nostr subscription for new app
97 createAppEventListener := &createAppConsumer{svc: svc, pool: pool}
98 svc.eventPublisher.RegisterSubscriber(createAppEventListener)
99
100 // register a subscriber for events of "nwc_app_updated" which handles re-publishing of nip47 event info
101 updateAppEventListener := &updateAppConsumer{svc: svc}
102 svc.eventPublisher.RegisterSubscriber(updateAppEventListener)
103
104 // start each app wallet subscription which have a child derived wallet key
105 svc.startAllExistingAppsWalletSubscriptions(ctx, pool)
106
107 // check if there are still legacy apps in DB
108 var legacyAppCount int64
109 result := svc.db.Model(&db.App{}).Where("wallet_pubkey IS NULL").Count(&legacyAppCount)
110 if result.Error != nil {
111 logger.Logger.WithError(result.Error).Error("Failed to count Legacy Apps")
112 }
113 if legacyAppCount > 0 {
114 go func() {
115 logger.Logger.WithField("legacy_app_count", legacyAppCount).Info("Starting legacy app subscription")
116 // legacy single wallet subscription - only subscribe once for all legacy apps
117 // to ensure we do not get duplicate events
118 svc.startAppWalletSubscription(ctx, pool, svc.keys.GetNostrPublicKey())
119 }()
120 }
121
122 go func() {
123 <-ctx.Done()
124 logger.Logger.Info("Main context cancelled, exiting...")
125
126 pool.Close("exiting")
127 logger.Logger.Info("Relay subroutine ended")
128
129 svc.eventPublisher.RemoveSubscriber(createAppEventListener)
130 svc.eventPublisher.RemoveSubscriber(updateAppEventListener)
131 }()
132
133 return nil
134 }
135
136 // In case the relay somehow loses events or the hub updates with
137 // new capabilities, we re-publish info events for all apps on startup
138 // to ensure that they are retrievable for all connections
139 func (svc *service) publishAllAppInfoEvents() {
140 func() {
141 var legacyAppCount int64
142 result := svc.db.Model(&db.App{}).Where("wallet_pubkey IS NULL").Count(&legacyAppCount)
143 if result.Error != nil {
144 logger.Logger.WithError(result.Error).Error("Failed to fetch App records with empty WalletPubkey")
145 return
146 }
147 if legacyAppCount > 0 {
148 logger.Logger.WithField("legacy_app_count", legacyAppCount).Debug("Enqueuing publish of legacy info event")
149 for _, relayUrl := range svc.cfg.GetRelayUrls() {
150 svc.nip47Service.EnqueueNip47InfoPublishRequest(0 /* unused */, svc.keys.GetNostrPublicKey(), svc.keys.GetNostrSecretKey(), relayUrl)
151 }
152 }
153 }()
154
155 var apps []db.App
156 result := svc.db.Where("wallet_pubkey IS NOT NULL").Find(&apps)
157 if result.Error != nil {
158 logger.Logger.WithError(result.Error).Error("Failed to fetch App records with non-empty WalletPubkey")
159 return
160 }
161
162 for _, app := range apps {
163 func(app db.App) {
164 // queue info event publish request for all existing apps
165 walletPrivKey, err := svc.keys.GetAppWalletKey(app.ID)
166 if err != nil {
167 logger.Logger.WithError(err).WithFields(logrus.Fields{
168 "app_id": app.ID}).Error("Could not get app wallet key")
169 return
170 }
171 logger.Logger.WithField("app_id", app.ID).Debug("Enqueuing publish of app info event")
172 for _, relayUrl := range svc.cfg.GetRelayUrls() {
173 svc.nip47Service.EnqueueNip47InfoPublishRequest(app.ID, *app.WalletPubkey, walletPrivKey, relayUrl)
174 }
175 }(app)
176 }
177 }
178
179 func (svc *service) startAllExistingAppsWalletSubscriptions(ctx context.Context, pool *nostr.SimplePool) {
180 var apps []db.App
181 result := svc.db.Where("wallet_pubkey IS NOT NULL").Find(&apps)
182 if result.Error != nil {
183 logger.Logger.WithError(result.Error).Error("Failed to fetch App records with non-empty WalletPubkey")
184 return
185 }
186
187 for _, app := range apps {
188 go func(app db.App) {
189 svc.startAppWalletSubscription(ctx, pool, *app.WalletPubkey)
190 }(app)
191 }
192 }
193
194 func (svc *service) startAppWalletSubscription(ctx context.Context, pool *nostr.SimplePool, appWalletPubKey string) error {
195
196 logger.Logger.Info("Subscribing to events for wallet ", appWalletPubKey)
197
198 filter := nostr.Filter{
199 Tags: nostr.TagMap{"p": []string{appWalletPubKey}},
200 Kinds: []int{models.REQUEST_KIND},
201 }
202
203 for {
204 subCtx, cancelSubscription := context.WithCancel(ctx)
205 eventsChannel := pool.SubscribeMany(subCtx, svc.cfg.GetRelayUrls(), filter)
206
207 // register a subscriber for "nwc_app_deleted" events, which handles
208 // cancelling the nostr subscription and nip47 info event deletion
209 deleteAppSubscriber := deleteAppConsumer{
210 cancelSubscription: cancelSubscription,
211 walletPubkey: appWalletPubKey,
212 svc: svc,
213 pool: pool,
214 }
215
216 svc.eventPublisher.RegisterSubscriber(&deleteAppSubscriber)
217
218 err := svc.watchSubscription(subCtx, pool, eventsChannel)
219
220 svc.eventPublisher.RemoveSubscriber(&deleteAppSubscriber)
221 if err != nil {
222 logger.Logger.WithError(err).Error("got an error from the relay while listening to subscription, resubscribing")
223 time.Sleep(3 * time.Second)
224 continue
225 }
226 break
227 }
228 return nil
229 }
230
231 func (svc *service) watchSubscription(ctx context.Context, pool *nostr.SimplePool, eventsChannel chan nostr.RelayEvent) error {
232 eventsChannelClosed := make(chan struct{})
233 go func() {
234 // loop through incoming events
235 for event := range eventsChannel {
236 select {
237 case <-ctx.Done():
238 return
239 default:
240 go svc.nip47Service.HandleEvent(ctx, pool, event.Event, svc.GetLNClient())
241 }
242 }
243 logger.Logger.Debug("Relay subscription events channel ended")
244 eventsChannelClosed <- struct{}{}
245 }()
246
247 select {
248 case <-ctx.Done():
249 logger.Logger.Info("Exiting subscription due to context exit...")
250 return nil
251 case <-eventsChannelClosed:
252 // in go-nostr pool, currently if the relay sends a close that is not "auth-required:"
253 // this will trigger closing the subscription channel. We return an error to trigger a resubscribe.
254 logger.Logger.Info("Subscription was exited abnormally")
255 return errors.New("subscription exited abnormally")
256 }
257 }
258
259 func (svc *service) StartApp(encryptionKey string) error {
260 defer func() {
261 svc.startupState = ""
262 }()
263
264 svc.startupState = "Initializing"
265 albyIdentifier, err := svc.albyOAuthSvc.GetUserIdentifier()
266 if err != nil {
267 return err
268 }
269 if albyIdentifier != "" && !svc.albyOAuthSvc.IsConnected(svc.ctx) {
270 return errors.New("alby account is not authenticated")
271 }
272
273 if svc.lnClient != nil {
274 return errors.New("app already started")
275 }
276 unlockPasswordCheckSet, err := svc.cfg.IsUnlockPasswordCheckSet()
277 if err != nil {
278 logger.Logger.WithError(err).Error("Failed to check unlock password check")
279 return fmt.Errorf("check unlock password check: %w", err)
280 }
281 if !unlockPasswordCheckSet {
282 logger.Logger.Error("Unlock password check is missing from the database")
283 return errors.New("your wallet data is incomplete and cannot be unlocked. Please restore from a backup")
284 }
285 if !svc.cfg.CheckUnlockPassword(encryptionKey) {
286 logger.Logger.Errorf("Invalid password")
287 return errors.New("invalid password")
288 }
289
290 err = svc.cfg.LoadJWTSecret(encryptionKey)
291 if err != nil {
292 logger.Logger.WithError(err).Error("Failed to load JWT secret")
293 return err
294 }
295
296 err = svc.keys.Init(svc.cfg, encryptionKey)
297 if err != nil {
298 logger.Logger.WithError(err).Error("Failed to init nostr keys")
299 return err
300 }
301
302 ctx, cancelFn := context.WithCancel(svc.ctx)
303
304 svc.startupState = "Launching Node"
305 err = svc.launchLNBackend(ctx, encryptionKey)
306 if err != nil {
307 logger.Logger.Errorf("Failed to launch LN backend: %v", err)
308 svc.eventPublisher.Publish(&events.Event{
309 Event: "nwc_node_start_failed",
310 })
311 cancelFn()
312 return err
313 }
314
315 svc.swapsService = swaps.NewSwapsService(ctx, svc.db, svc.cfg, svc.keys, svc.eventPublisher, svc.GetLNClient(), svc.transactionsService, encryptionKey)
316
317 svc.publishAllAppInfoEvents()
318
319 svc.startupState = "Connecting To Relay"
320 err = svc.startNostr(ctx)
321 if err != nil {
322 cancelFn()
323 return err
324 }
325
326 svc.appCancelFn = cancelFn
327
328 return nil
329 }
330
331 func (svc *service) launchLNBackend(ctx context.Context, encryptionKey string) error {
332 if svc.lnClient != nil {
333 logger.Logger.Error("LNClient already started")
334 return errors.New("LNClient already started")
335 }
336
337 svc.wg.Add(1)
338 go func() {
339 // ensure the LNClient is stopped properly before exiting
340 <-ctx.Done()
341 svc.stopLNClient()
342 }()
343
344 lnBackend, _ := svc.cfg.Get("LNBackendType", "")
345 if lnBackend == "" {
346 return errors.New("no LNBackendType specified")
347 }
348
349 logger.Logger.Infof("Launching LN Backend: %s", lnBackend)
350 var lnClient lnclient.LNClient
351 var err error
352 vssEnabled := false
353 switch lnBackend {
354 case config.LNDBackendType:
355 LNDAddress, _ := svc.cfg.Get("LNDAddress", encryptionKey)
356 LNDCertHex, _ := svc.cfg.Get("LNDCertHex", encryptionKey)
357 LNDMacaroonHex, _ := svc.cfg.Get("LNDMacaroonHex", encryptionKey)
358 lnClient, err = lnd.NewLNDService(ctx, svc.eventPublisher, LNDAddress, LNDCertHex, LNDMacaroonHex)
359 case config.LDKBackendType:
360 mnemonic, _ := svc.cfg.Get("Mnemonic", encryptionKey)
361 ldkWorkdir := path.Join(svc.cfg.GetEnv().Workdir, "ldk")
362 var vssToken string
363 vssToken, err = svc.requestVssToken(ctx)
364 if err != nil {
365 logger.Logger.WithError(err).Error("Failed to request VSS token")
366 return err
367 }
368 vssEnabled = vssToken != ""
369
370 svc.startupState = "Launching Node"
371 setStartupState := func(startupState string) {
372 svc.startupState = startupState
373 }
374
375 channelPeerSuggestions, suggestionsErr := svc.albySvc.GetChannelPeerSuggestions(ctx)
376 if suggestionsErr != nil {
377 logger.Logger.WithError(suggestionsErr).Warn("Failed to fetch channel peer suggestions for LSPS2 liquidity source")
378 }
379 lnClient, err = ldk.NewLDKService(ctx, svc.cfg, svc.eventPublisher, mnemonic, ldkWorkdir, vssToken, setStartupState, channelPeerSuggestions)
380 case config.PhoenixBackendType:
381 PhoenixdAddress, _ := svc.cfg.Get("PhoenixdAddress", encryptionKey)
382 PhoenixdAuthorization, _ := svc.cfg.Get("PhoenixdAuthorization", encryptionKey)
383
384 lnClient, err = phoenixd.NewPhoenixService(ctx, PhoenixdAddress, PhoenixdAuthorization)
385 case config.CashuBackendType:
386 mnemonic, _ := svc.cfg.Get("Mnemonic", encryptionKey)
387 cashuMintUrl, _ := svc.cfg.Get("CashuMintUrl", encryptionKey)
388 cashuWorkdir := path.Join(svc.cfg.GetEnv().Workdir, "cashu")
389
390 lnClient, err = cashu.NewCashuService(svc.cfg, cashuWorkdir, mnemonic, cashuMintUrl)
391 case config.BarkBackendType:
392 mnemonic, _ := svc.cfg.Get("Mnemonic", encryptionKey)
393 env := svc.cfg.GetEnv()
394 barkWorkdir := path.Join(env.Workdir, "bark")
395
396 lnClient, err = bark.NewBarkService(ctx, svc.eventPublisher, barkWorkdir, mnemonic, bark.Config{
397 Network: svc.cfg.GetNetwork(),
398 ServerAddress: env.BarkServer,
399 EsploraAddress: env.BarkEsploraServer,
400 ServerAccessToken: env.BarkServerAccessToken,
401 LogLevel: env.BarkLogLevel,
402 LogToFile: env.LogToFile,
403 })
404 case config.CLNBackendType:
405 CLNAddress, _ := svc.cfg.Get("CLNAddress", encryptionKey)
406 CLNLightningDir, _ := svc.cfg.Get("CLNLightningDir", encryptionKey)
407 CLNAddressHold, _ := svc.cfg.Get("CLNAddressHold", encryptionKey)
408 lnClient, err = cln.NewCLNService(ctx, svc.eventPublisher, CLNAddress, CLNLightningDir, CLNAddressHold)
409 default:
410 logger.Logger.WithField("backend_type", lnBackend).Error("Unsupported LNBackendType")
411 return fmt.Errorf("unsupported backend type: %s", lnBackend)
412 }
413 if err != nil {
414 logger.Logger.WithError(err).Error("Failed to launch LN backend")
415 return err
416 }
417
418 // TODO: call a method on the LNClient here to check the LNClient is actually connectable,
419 // (e.g. lnClient.CheckConnection()) Rather than it being a side-effect
420 // in the LNClient init function
421
422 svc.lnClient = lnClient
423 info, err := lnClient.GetInfo(ctx)
424 if err != nil {
425 logger.Logger.WithError(err).Error("Failed to fetch node info")
426 }
427 if info != nil {
428 svc.eventPublisher.SetGlobalProperty("node_id", info.Pubkey)
429 svc.eventPublisher.SetGlobalProperty("network", info.Network)
430 }
431
432 // Mark that the node has successfully started
433 // This will ensure the user cannot go through the setup again
434 err = svc.cfg.SetUpdate("NodeLastStartTime", strconv.FormatInt(time.Now().Unix(), 10), "")
435 if err != nil {
436 logger.Logger.WithError(err).Error("Failed to set last node start time")
437 }
438
439 svc.eventPublisher.Publish(&events.Event{
440 Event: "nwc_node_started",
441 Properties: map[string]interface{}{
442 "node_type": lnBackend,
443 "vss_enabled": vssEnabled,
444 },
445 })
446
447 return nil
448 }
449
450 func (svc *service) requestVssToken(ctx context.Context) (string, error) {
451 nodeLastStartTime, _ := svc.cfg.Get("NodeLastStartTime", "")
452
453 // for brand new nodes, consider enabling VSS
454 if nodeLastStartTime == "" && svc.cfg.GetEnv().LDKVssUrl != "" {
455 svc.startupState = "Checking Subscription"
456 albyUserIdentifier, err := svc.albyOAuthSvc.GetUserIdentifier()
457 if err != nil {
458 logger.Logger.WithError(err).Error("Failed to fetch alby user identifier")
459 return "", err
460 }
461 if albyUserIdentifier != "" {
462 me, err := svc.albyOAuthSvc.GetMe(ctx)
463 if err != nil {
464 logger.Logger.WithError(err).Error("Failed to fetch alby user")
465 return "", err
466 }
467 // only activate VSS for Alby paid subscribers
468 if me.Subscription.PlanCode != "" {
469 svc.cfg.SetUpdate("LdkVssEnabled", "true", "")
470 }
471 }
472 }
473
474 vssToken := ""
475 vssEnabled, _ := svc.cfg.Get("LdkVssEnabled", "")
476 if vssEnabled == "true" {
477 svc.startupState = "Fetching VSS token"
478 vssNodeIdentifier, err := ldk.GetVssNodeIdentifier(svc.keys)
479 if err != nil {
480 logger.Logger.WithError(err).Error("Failed to get VSS node identifier")
481 return "", err
482 }
483 vssToken, err = svc.albyOAuthSvc.GetVssAuthToken(ctx, vssNodeIdentifier)
484 if err != nil {
485 logger.Logger.WithError(err).Error("Failed to fetch VSS JWT token")
486
487 existingVssToken, _ := svc.cfg.Get("VssToken", "")
488 if existingVssToken != "" {
489 logger.Logger.Warn("Using stored VSS JWT token")
490 return existingVssToken, nil
491 }
492
493 return "", err
494 }
495 err = svc.cfg.SetUpdate("VssToken", vssToken, "")
496 if err != nil {
497 logger.Logger.WithError(err).Error("Failed to save VSS JWT token to user config")
498 }
499 }
500 return vssToken, nil
501 }
502