service.go raw
1 package service
2
3 import (
4 "context"
5 "os"
6 "path/filepath"
7 "strings"
8 "sync"
9 "sync/atomic"
10 "time"
11
12 "github.com/adrg/xdg"
13 "github.com/sirupsen/logrus"
14 "gorm.io/gorm"
15
16 "github.com/joho/godotenv"
17 "github.com/kelseyhightower/envconfig"
18
19 "github.com/getAlby/hub/alby"
20 "github.com/getAlby/hub/events"
21 "github.com/getAlby/hub/logger"
22 "github.com/getAlby/hub/service/keys"
23 "github.com/getAlby/hub/swaps"
24 "github.com/getAlby/hub/transactions"
25 "github.com/getAlby/hub/version"
26
27 "github.com/getAlby/hub/config"
28 "github.com/getAlby/hub/db"
29 "github.com/getAlby/hub/lnclient"
30 "github.com/getAlby/hub/nip47"
31 )
32
33 type service struct {
34 cfg config.Config
35
36 db *gorm.DB
37 lnClient lnclient.LNClient
38 lnClientShuttingDown atomic.Bool
39 transactionsService transactions.TransactionsService
40 swapsService swaps.SwapsService
41 albySvc alby.AlbyService
42 albyOAuthSvc alby.AlbyOAuthService
43 eventPublisher events.EventPublisher
44 ctx context.Context
45 wg *sync.WaitGroup
46 nip47Service nip47.Nip47Service
47 appCancelFn context.CancelFunc
48 keys keys.Keys
49 relayStatuses []RelayStatus
50 startupState string
51 }
52
53 func NewService(ctx context.Context) (*service, error) {
54 // Load config from environment variables / .GetEnv() file
55 godotenv.Load(".env")
56 appConfig := &config.AppConfig{}
57 err := envconfig.Process("", appConfig)
58 if err != nil {
59 return nil, err
60 }
61
62 logger.Init(appConfig.LogLevel)
63 logger.Logger.Info("AlbyHub " + version.Tag)
64
65 if appConfig.Workdir == "" {
66 appConfig.Workdir = filepath.Join(xdg.DataHome, "/albyhub")
67 logger.Logger.WithField("workdir", appConfig.Workdir).Info("No workdir specified, using default")
68 }
69 // make sure workdir exists
70 os.MkdirAll(appConfig.Workdir, os.ModePerm)
71
72 if appConfig.LogToFile {
73 err = logger.AddFileLogger(appConfig.Workdir)
74 if err != nil {
75 return nil, err
76 }
77 }
78
79 err = finishRestoreNode(appConfig.Workdir)
80 if err != nil {
81 logger.Logger.WithError(err).Error("failed to restore backup")
82 return nil, err
83 }
84
85 // If DATABASE_URI is a URI or a path, leave it unchanged.
86 // If it only contains a filename, prepend the workdir.
87 if !strings.HasPrefix(appConfig.DatabaseUri, "file:") {
88 databasePath, _ := filepath.Split(appConfig.DatabaseUri)
89 if databasePath == "" {
90 appConfig.DatabaseUri = filepath.Join(appConfig.Workdir, appConfig.DatabaseUri)
91 }
92 }
93
94 gormDB, err := db.NewDB(appConfig.DatabaseUri, appConfig.LogDBQueries)
95 if err != nil {
96 return nil, err
97 }
98
99 cfg, err := config.NewConfig(appConfig, gormDB)
100 if err != nil {
101 return nil, err
102 }
103
104 // write auto unlock password from env to user config
105 if appConfig.AutoUnlockPassword != "" {
106 err = cfg.SetUpdate("AutoUnlockPassword", appConfig.AutoUnlockPassword, "")
107 if err != nil {
108 return nil, err
109 }
110 }
111 autoUnlockPassword, err := cfg.Get("AutoUnlockPassword", "")
112 if err != nil {
113 return nil, err
114 }
115
116 eventPublisher := events.NewEventPublisher()
117
118 keys := keys.NewKeys()
119
120 albySvc := alby.NewAlbyService()
121 albyOAuthSvc := alby.NewAlbyOAuthService(gormDB, cfg, keys, eventPublisher)
122
123 transactionsSvc := transactions.NewTransactionsService(gormDB, eventPublisher)
124
125 var wg sync.WaitGroup
126 svc := &service{
127 cfg: cfg,
128 ctx: ctx,
129 wg: &wg,
130 eventPublisher: eventPublisher,
131 albySvc: albySvc,
132 albyOAuthSvc: albyOAuthSvc,
133 nip47Service: nip47.NewNip47Service(gormDB, cfg, keys, eventPublisher, albyOAuthSvc),
134 transactionsService: transactionsSvc,
135 db: gormDB,
136 keys: keys,
137 }
138
139 eventPublisher.RegisterSubscriber(svc.transactionsService)
140 eventPublisher.RegisterSubscriber(svc.nip47Service)
141 eventPublisher.RegisterSubscriber(svc.albyOAuthSvc)
142 eventPublisher.RegisterSubscriber(&paymentForwardedConsumer{
143 db: gormDB,
144 })
145
146 eventPublisher.Publish(&events.Event{
147 Event: "nwc_started",
148 Properties: map[string]interface{}{
149 "version": version.Tag,
150 },
151 })
152
153 if appConfig.GoProfilerAddr != "" {
154 startProfiler(ctx, appConfig.GoProfilerAddr)
155 }
156
157 if autoUnlockPassword != "" {
158 nodeLastStartTime, _ := cfg.Get("NodeLastStartTime", "")
159 if nodeLastStartTime != "" {
160 svc.StartApp(autoUnlockPassword)
161 }
162 }
163
164 go func() {
165 for {
166 select {
167 case <-ctx.Done():
168 return
169 default:
170 time.Sleep(10 * time.Minute)
171 svc.removeExcessEvents()
172 }
173 }
174 }()
175
176 return svc, nil
177 }
178
179 func (svc *service) noticeHandler(notice string) {
180 logger.Logger.Infof("Received a notice %s", notice)
181 }
182
183 func finishRestoreNode(workDir string) error {
184 restoreDir := filepath.Join(workDir, "restore")
185 if restoreDirStat, err := os.Stat(restoreDir); err == nil && restoreDirStat.IsDir() {
186 logger.Logger.WithField("restoreDir", restoreDir).Infof("Restore directory found. Finishing Node restore")
187
188 existingFiles, err := os.ReadDir(restoreDir)
189 if err != nil {
190 logger.Logger.WithError(err).Error("Failed to read WORK_DIR")
191 return err
192 }
193
194 for _, file := range existingFiles {
195 if file.Name() != "restore" {
196 err = os.RemoveAll(filepath.Join(workDir, file.Name()))
197 if err != nil {
198 logger.Logger.WithField("filename", file.Name()).WithError(err).Error("Failed to remove file")
199 return err
200 }
201 logger.Logger.WithField("filename", file.Name()).Info("removed file")
202 }
203 }
204
205 files, err := os.ReadDir(restoreDir)
206 if err != nil {
207 logger.Logger.WithError(err).Error("Failed to read restore directory")
208 return err
209 }
210 for _, file := range files {
211 err = os.Rename(filepath.Join(restoreDir, file.Name()), filepath.Join(workDir, file.Name()))
212 if err != nil {
213 logger.Logger.WithField("filename", file.Name()).WithError(err).Error("Failed to move file")
214 return err
215 }
216 logger.Logger.WithField("filename", file.Name()).Info("copied file from restore directory")
217 }
218 err = os.RemoveAll(restoreDir)
219 if err != nil {
220 logger.Logger.WithError(err).Error("Failed to remove restore directory")
221 return err
222 }
223 logger.Logger.WithField("restoreDir", restoreDir).Info("removed restore directory")
224 }
225 return nil
226 }
227
228 func (svc *service) Shutdown() {
229 svc.StopApp()
230 svc.eventPublisher.PublishSync(&events.Event{
231 Event: "nwc_stopped",
232 })
233 db.Stop(svc.db)
234 }
235
236 func (svc *service) GetDB() *gorm.DB {
237 return svc.db
238 }
239
240 func (svc *service) GetConfig() config.Config {
241 return svc.cfg
242 }
243
244 func (svc *service) GetAlbySvc() alby.AlbyService {
245 return svc.albySvc
246 }
247
248 func (svc *service) GetAlbyOAuthSvc() alby.AlbyOAuthService {
249 return svc.albyOAuthSvc
250 }
251
252 func (svc *service) GetNip47Service() nip47.Nip47Service {
253 return svc.nip47Service
254 }
255
256 func (svc *service) GetEventPublisher() events.EventPublisher {
257 return svc.eventPublisher
258 }
259
260 func (svc *service) GetLNClient() lnclient.LNClient {
261 if svc.lnClientShuttingDown.Load() {
262 return nil
263 }
264 return svc.lnClient
265 }
266
267 func (svc *service) GetTransactionsService() transactions.TransactionsService {
268 return svc.transactionsService
269 }
270
271 func (svc *service) GetSwapsService() swaps.SwapsService {
272 return svc.swapsService
273 }
274
275 func (svc *service) GetKeys() keys.Keys {
276 return svc.keys
277 }
278
279 func (svc *service) GetRelayStatuses() []RelayStatus {
280 return svc.relayStatuses
281 }
282
283 func (svc *service) GetStartupState() string {
284 return svc.startupState
285 }
286
287 func (svc *service) removeExcessEvents() {
288 logger.Logger.Debug("Cleaning up excess events")
289
290 maxEvents := 1000
291 // estimated less than 1 second to delete, it should not lock the DB
292 maxEventsToDelete := 5000
293 // if we only have a few excess events, don't run the task
294 minEventsToDelete := 100
295
296 var events []db.RequestEvent
297 err := svc.db.Select("id").Order("id asc").Limit(maxEvents + maxEventsToDelete).Find(&events).Error
298 if err != nil {
299 logger.Logger.WithError(err).Error("Failed to fetch request events")
300 }
301
302 numEventsToDelete := len(events) - maxEvents
303
304 if numEventsToDelete < minEventsToDelete {
305 return
306 }
307 deleteEventsBelowId := events[numEventsToDelete].ID
308
309 logger.Logger.WithFields(logrus.Fields{
310 "amount": numEventsToDelete,
311 "below_id": deleteEventsBelowId,
312 }).Debug("Removing excess events")
313
314 startTime := time.Now()
315 err = svc.db.Exec("delete from request_events where id < ?", deleteEventsBelowId).Error
316 if err != nil {
317 logger.Logger.WithError(err).WithFields(logrus.Fields{
318 "amount": numEventsToDelete,
319 "below_id": deleteEventsBelowId,
320 }).Error("Failed to delete excess request events")
321 return
322 }
323 logger.Logger.WithFields(logrus.Fields{
324 "amount": numEventsToDelete,
325 "below_id": deleteEventsBelowId,
326 "duration_seconds": time.Since(startTime).Seconds(),
327 }).Info("Removed excess events")
328
329 // TODO: REMOVE AFTER 2026-01-01
330 // this is needed due to cascading delete previously not working
331 err = svc.db.Exec("delete from response_events where request_id < ?", deleteEventsBelowId).Error
332 if err != nil {
333 logger.Logger.WithError(err).WithFields(logrus.Fields{
334 "amount": numEventsToDelete,
335 "below_id": deleteEventsBelowId,
336 }).Error("Failed to delete excess response events")
337 return
338 }
339 }
340