mirror of
https://github.com/getAlby/hub.git
synced 2026-08-13 12:33:39 +02:00
feat: support multiple relays (#1802)
* feat: support multiple relays (WIP) * fix: multiple NWC url construction for multiple relays * fix: startup * chore: update go-nostr to fix pool relay reconnect * fix: split nip-47 queue info publish event for each relay url * fix: add relayUrls to dispatched nwc success events * fix: ensure newly created app info publishing is not blocked by sleep from failed publishes * feat: multiple relays improvements - update app deleted and updated consumers to handle multiple relays - handle multiple relays in healthcheck - display relay online status in about page - renaming and improved comments * chore: simplify error messages * fix: unnecessary db error log when checking if app has notification permission * fix: tests * chore: remove old comment * fix: app wallet subscription context usage
This commit is contained in:
Roland
GitHub
parent
944bb11c28
commit
377a3169c6
28 changed files with 487 additions and 403 deletions
|
|
@ -153,7 +153,7 @@ For more information on the Go pprof library, see the [official documentation](h
|
|||
|
||||
The following configuration options can be set as environment variables or in a .env file
|
||||
|
||||
- `RELAY`: default: "wss://relay.getalby.com/v1"
|
||||
- `RELAY`: default: "wss://relay.getalby.com/v1" (can support multiple separated by commas)
|
||||
- `JWT_SECRET`: A randomly generated secret string, applied if no JWT secret is already set. (only needed in http mode). If not provided, one will be automatically generated. On password change, a new JWT secret will be generated.
|
||||
- `DATABASE_URI`: A sqlite filename or postgres URL. Default is SQLite DB `nwc.db` without a path, which will be put in the user home directory: $XDG_DATA_HOME/albyhub/nwc.db
|
||||
- `PORT`: The port on which the app should listen on (default: 8080)
|
||||
|
|
|
|||
|
|
@ -933,7 +933,7 @@ func (svc *albyOAuthService) activateAlbyAccountNWCNode(ctx context.Context, wal
|
|||
|
||||
activateNodeRequest := activateNWCNodeRequest{
|
||||
WalletPubkey: walletServicePubkey,
|
||||
RelayUrl: svc.cfg.GetRelayUrl(),
|
||||
RelayUrl: svc.cfg.GetRelayUrls()[0], // TODO: pass all URLs if/when Alby Account supports it
|
||||
}
|
||||
|
||||
body := bytes.NewBuffer([]byte{})
|
||||
|
|
|
|||
38
api/api.go
38
api/api.go
|
|
@ -102,7 +102,7 @@ func (api *api) CreateApp(createAppRequest *CreateAppRequest) (*CreateAppRespons
|
|||
return nil, err
|
||||
}
|
||||
|
||||
relayUrl := api.cfg.GetRelayUrl()
|
||||
relayUrls := api.cfg.GetRelayUrls()
|
||||
|
||||
lightningAddress, err := api.albyOAuthSvc.GetLightningAddress()
|
||||
if err != nil {
|
||||
|
|
@ -115,14 +115,16 @@ func (api *api) CreateApp(createAppRequest *CreateAppRequest) (*CreateAppRespons
|
|||
responseBody.Pubkey = app.AppPubkey
|
||||
responseBody.PairingSecret = pairingSecretKey
|
||||
responseBody.WalletPubkey = *app.WalletPubkey
|
||||
responseBody.RelayUrl = relayUrl
|
||||
responseBody.RelayUrls = relayUrls
|
||||
responseBody.Lud16 = lightningAddress
|
||||
|
||||
if createAppRequest.ReturnTo != "" {
|
||||
returnToUrl, err := url.Parse(createAppRequest.ReturnTo)
|
||||
if err == nil {
|
||||
query := returnToUrl.Query()
|
||||
query.Add("relay", relayUrl)
|
||||
for _, relayUrl := range relayUrls {
|
||||
query.Add("relay", relayUrl)
|
||||
}
|
||||
query.Add("pubkey", *app.WalletPubkey)
|
||||
if lightningAddress != "" && !app.Isolated {
|
||||
query.Add("lud16", lightningAddress)
|
||||
|
|
@ -136,7 +138,7 @@ func (api *api) CreateApp(createAppRequest *CreateAppRequest) (*CreateAppRespons
|
|||
if lightningAddress != "" && !app.Isolated {
|
||||
lud16 = fmt.Sprintf("&lud16=%s", lightningAddress)
|
||||
}
|
||||
responseBody.PairingUri = fmt.Sprintf("nostr+walletconnect://%s?relay=%s&secret=%s%s", *app.WalletPubkey, relayUrl, pairingSecretKey, lud16)
|
||||
responseBody.PairingUri = fmt.Sprintf("nostr+walletconnect://%s?relay=%s&secret=%s%s", *app.WalletPubkey, strings.Join(relayUrls, "&relay="), pairingSecretKey, lud16)
|
||||
|
||||
return responseBody, nil
|
||||
}
|
||||
|
|
@ -1203,15 +1205,24 @@ func (api *api) GetInfo(ctx context.Context) (*InfoResponse, error) {
|
|||
info.VssSupported = backendType == config.LDKBackendType && api.cfg.GetEnv().LDKVssUrl != ""
|
||||
info.AutoUnlockPasswordEnabled = autoUnlockPassword != ""
|
||||
info.AutoUnlockPasswordSupported = api.cfg.GetEnv().IsDefaultClientId()
|
||||
albyUserIdentifier, err := api.albyOAuthSvc.GetUserIdentifier()
|
||||
info.Relays = []InfoResponseRelay{}
|
||||
for _, relayStatus := range api.svc.GetRelayStatuses() {
|
||||
info.Relays = append(info.Relays, InfoResponseRelay{
|
||||
Url: relayStatus.Url,
|
||||
Online: relayStatus.Online,
|
||||
})
|
||||
}
|
||||
|
||||
info.MempoolUrl = api.cfg.GetMempoolUrl()
|
||||
info.Relay = api.cfg.GetRelayUrl()
|
||||
info.AlbyAccountConnected = api.albyOAuthSvc.IsConnected(ctx)
|
||||
|
||||
albyUserIdentifier, err := api.albyOAuthSvc.GetUserIdentifier()
|
||||
if err != nil {
|
||||
logger.Logger.WithError(err).Error("Failed to get alby user identifier")
|
||||
return nil, err
|
||||
}
|
||||
info.AlbyUserIdentifier = albyUserIdentifier
|
||||
info.AlbyAccountConnected = api.albyOAuthSvc.IsConnected(ctx)
|
||||
|
||||
if api.svc.GetLNClient() != nil {
|
||||
nodeInfo, err := api.svc.GetLNClient().GetInfo(ctx)
|
||||
if err != nil {
|
||||
|
|
@ -1532,9 +1543,16 @@ func (api *api) Health(ctx context.Context) (*HealthResponse, error) {
|
|||
alarms = append(alarms, NewHealthAlarm(HealthAlarmKindAlbyService, albyInfo.Incidents))
|
||||
}
|
||||
|
||||
isNostrRelayReady := api.svc.IsRelayReady()
|
||||
if !isNostrRelayReady {
|
||||
alarms = append(alarms, NewHealthAlarm(HealthAlarmKindNostrRelayOffline, nil))
|
||||
isAnyNostrRelayOffline := len(api.svc.GetRelayStatuses()) == 0
|
||||
offlineRelayUrls := []string{}
|
||||
for _, relayStatus := range api.svc.GetRelayStatuses() {
|
||||
if !relayStatus.Online {
|
||||
isAnyNostrRelayOffline = true
|
||||
offlineRelayUrls = append(offlineRelayUrls, relayStatus.Url)
|
||||
}
|
||||
}
|
||||
if isAnyNostrRelayOffline {
|
||||
alarms = append(alarms, NewHealthAlarm(HealthAlarmKindNostrRelayOffline, offlineRelayUrls))
|
||||
}
|
||||
|
||||
ldkVssEnabled, _ := api.cfg.Get("LdkVssEnabled", "")
|
||||
|
|
|
|||
|
|
@ -250,45 +250,50 @@ type SetupRequest struct {
|
|||
}
|
||||
|
||||
type CreateAppResponse struct {
|
||||
PairingUri string `json:"pairingUri"`
|
||||
PairingSecret string `json:"pairingSecretKey"`
|
||||
Pubkey string `json:"pairingPublicKey"`
|
||||
RelayUrl string `json:"relayUrl"`
|
||||
WalletPubkey string `json:"walletPubkey"`
|
||||
Lud16 string `json:"lud16"`
|
||||
Id uint `json:"id"`
|
||||
Name string `json:"name"`
|
||||
ReturnTo string `json:"returnTo"`
|
||||
PairingUri string `json:"pairingUri"`
|
||||
PairingSecret string `json:"pairingSecretKey"`
|
||||
Pubkey string `json:"pairingPublicKey"`
|
||||
RelayUrls []string `json:"relayUrls"`
|
||||
WalletPubkey string `json:"walletPubkey"`
|
||||
Lud16 string `json:"lud16"`
|
||||
Id uint `json:"id"`
|
||||
Name string `json:"name"`
|
||||
ReturnTo string `json:"returnTo"`
|
||||
}
|
||||
|
||||
type User struct {
|
||||
Email string `json:"email"`
|
||||
}
|
||||
|
||||
type InfoResponseRelay struct {
|
||||
Url string `json:"url"`
|
||||
Online bool `json:"online"`
|
||||
}
|
||||
|
||||
type InfoResponse struct {
|
||||
BackendType string `json:"backendType"`
|
||||
SetupCompleted bool `json:"setupCompleted"`
|
||||
OAuthRedirect bool `json:"oauthRedirect"`
|
||||
Running bool `json:"running"`
|
||||
Unlocked bool `json:"unlocked"`
|
||||
AlbyAuthUrl string `json:"albyAuthUrl"`
|
||||
NextBackupReminder string `json:"nextBackupReminder"`
|
||||
AlbyUserIdentifier string `json:"albyUserIdentifier"`
|
||||
AlbyAccountConnected bool `json:"albyAccountConnected"`
|
||||
Version string `json:"version"`
|
||||
Network string `json:"network"`
|
||||
EnableAdvancedSetup bool `json:"enableAdvancedSetup"`
|
||||
LdkVssEnabled bool `json:"ldkVssEnabled"`
|
||||
VssSupported bool `json:"vssSupported"`
|
||||
StartupState string `json:"startupState"`
|
||||
StartupError string `json:"startupError"`
|
||||
StartupErrorTime time.Time `json:"startupErrorTime"`
|
||||
AutoUnlockPasswordSupported bool `json:"autoUnlockPasswordSupported"`
|
||||
AutoUnlockPasswordEnabled bool `json:"autoUnlockPasswordEnabled"`
|
||||
Currency string `json:"currency"`
|
||||
Relay string `json:"relay"`
|
||||
NodeAlias string `json:"nodeAlias"`
|
||||
MempoolUrl string `json:"mempoolUrl"`
|
||||
BackendType string `json:"backendType"`
|
||||
SetupCompleted bool `json:"setupCompleted"`
|
||||
OAuthRedirect bool `json:"oauthRedirect"`
|
||||
Running bool `json:"running"`
|
||||
Unlocked bool `json:"unlocked"`
|
||||
AlbyAuthUrl string `json:"albyAuthUrl"`
|
||||
NextBackupReminder string `json:"nextBackupReminder"`
|
||||
AlbyUserIdentifier string `json:"albyUserIdentifier"`
|
||||
AlbyAccountConnected bool `json:"albyAccountConnected"`
|
||||
Version string `json:"version"`
|
||||
Network string `json:"network"`
|
||||
EnableAdvancedSetup bool `json:"enableAdvancedSetup"`
|
||||
LdkVssEnabled bool `json:"ldkVssEnabled"`
|
||||
VssSupported bool `json:"vssSupported"`
|
||||
StartupState string `json:"startupState"`
|
||||
StartupError string `json:"startupError"`
|
||||
StartupErrorTime time.Time `json:"startupErrorTime"`
|
||||
AutoUnlockPasswordSupported bool `json:"autoUnlockPasswordSupported"`
|
||||
AutoUnlockPasswordEnabled bool `json:"autoUnlockPasswordEnabled"`
|
||||
Currency string `json:"currency"`
|
||||
Relays []InfoResponseRelay `json:"relays"`
|
||||
NodeAlias string `json:"nodeAlias"`
|
||||
MempoolUrl string `json:"mempoolUrl"`
|
||||
}
|
||||
|
||||
type UpdateSettingsRequest struct {
|
||||
|
|
|
|||
|
|
@ -151,9 +151,9 @@ func (cfg *config) GetJWTSecret() string {
|
|||
return secret
|
||||
}
|
||||
|
||||
func (cfg *config) GetRelayUrl() string {
|
||||
relayUrl, _ := cfg.Get("Relay", "")
|
||||
return relayUrl
|
||||
func (cfg *config) GetRelayUrls() []string {
|
||||
relayUrls, _ := cfg.Get("Relay", "")
|
||||
return strings.Split(relayUrls, ",")
|
||||
}
|
||||
|
||||
func (cfg *config) GetNetwork() string {
|
||||
|
|
|
|||
|
|
@ -77,7 +77,7 @@ type Config interface {
|
|||
SetIgnore(key string, value string, encryptionKey string) error
|
||||
SetUpdate(key string, value string, encryptionKey string) error
|
||||
GetJWTSecret() string
|
||||
GetRelayUrl() string
|
||||
GetRelayUrls() []string
|
||||
GetNetwork() string
|
||||
GetMempoolUrl() string
|
||||
GetEnv() *AppConfig
|
||||
|
|
|
|||
|
|
@ -34,7 +34,10 @@ export function HealthCheckAlert() {
|
|||
case "node_not_ready":
|
||||
return "Node is not ready";
|
||||
case "nostr_relay_offline":
|
||||
return "Could not connect to relay";
|
||||
return (
|
||||
"Could not connect to relay: " +
|
||||
(alarm.rawDetails as string[]).join(", ")
|
||||
);
|
||||
case "vss_no_subscription":
|
||||
return "Your lightning channel data is stored encrypted by Alby's Versioned Storage Service which is a paid feature. Restart your subscription or send your funds to another wallet as soon as possible.";
|
||||
}
|
||||
|
|
|
|||
|
|
@ -264,7 +264,8 @@ const NewAppInternal = ({ capabilities }: NewAppInternalProps) => {
|
|||
// this gives those apps the chance to know the user has enabled the connection
|
||||
const nwcEvent = new CustomEvent("nwc:success", {
|
||||
detail: {
|
||||
relayUrl: createAppResponse.relayUrl,
|
||||
relayUrl: createAppResponse.relayUrls[0], // TODO: deprecate
|
||||
relayUrls: createAppResponse.relayUrls, // TODO: add to spec
|
||||
walletPubkey: createAppResponse.walletPubkey,
|
||||
lud16: createAppResponse.lud16,
|
||||
},
|
||||
|
|
@ -276,7 +277,8 @@ const NewAppInternal = ({ capabilities }: NewAppInternalProps) => {
|
|||
window.opener.postMessage(
|
||||
{
|
||||
type: "nwc:success",
|
||||
relayUrl: createAppResponse.relayUrl,
|
||||
relayUrl: createAppResponse.relayUrls[0], // TODO: deprecate
|
||||
relayUrls: createAppResponse.relayUrls, // TODO: add to spec
|
||||
walletPubkey: createAppResponse.walletPubkey,
|
||||
lud16: createAppResponse.lud16,
|
||||
},
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
import Loading from "src/components/Loading";
|
||||
import SettingsHeader from "src/components/SettingsHeader";
|
||||
import { Badge } from "src/components/ui/badge";
|
||||
import { useAlbyMe } from "src/hooks/useAlbyMe";
|
||||
|
||||
import { useInfo } from "src/hooks/useInfo";
|
||||
|
|
@ -41,10 +42,15 @@ export function About() {
|
|||
</p>
|
||||
</div>
|
||||
<div className="grid gap-2">
|
||||
<p className="font-medium text-sm">Nostr Relay</p>
|
||||
<p className="text-muted-foreground text-sm slashed-zero">
|
||||
{info.relay}
|
||||
</p>
|
||||
<p className="font-medium text-sm">Nostr Relays</p>
|
||||
{info.relays.map((relay) => (
|
||||
<p className="flex items-center gap-2 text-muted-foreground text-sm">
|
||||
{relay.url}
|
||||
<Badge variant={relay.online ? "positive" : "destructive"}>
|
||||
{relay.online ? "online" : "offline"}
|
||||
</Badge>
|
||||
</p>
|
||||
))}
|
||||
</div>
|
||||
{info.albyAccountConnected && albyMe && (
|
||||
<div className="grid gap-2">
|
||||
|
|
|
|||
|
|
@ -152,7 +152,7 @@ export interface InfoResponse {
|
|||
albyUserIdentifier: string;
|
||||
network?: Network;
|
||||
version: string;
|
||||
relay: string;
|
||||
relays: { url: string; online: boolean }[];
|
||||
unlocked: boolean;
|
||||
enableAdvancedSetup: boolean;
|
||||
startupState: string;
|
||||
|
|
@ -269,7 +269,7 @@ export interface CreateAppResponse {
|
|||
pairingUri: string;
|
||||
pairingPublicKey: string;
|
||||
pairingSecretKey: string;
|
||||
relayUrl: string;
|
||||
relayUrls: string[];
|
||||
walletPubkey: string;
|
||||
lud16: string;
|
||||
returnTo: string;
|
||||
|
|
|
|||
|
|
@ -271,7 +271,7 @@ func TestCreateApp_FullPermission(t *testing.T) {
|
|||
mockConfig.On("GetEnv").Return(&config.AppConfig{})
|
||||
mockConfig.On("CheckUnlockPassword", "123").Return(true)
|
||||
mockConfig.On("GetJWTSecret").Return("dummy secret")
|
||||
mockConfig.On("GetRelayUrl").Return("")
|
||||
mockConfig.On("GetRelayUrls").Return([]string{})
|
||||
|
||||
mockKeys := mocks.NewMockKeys(t)
|
||||
mockKeys.On("GetAppWalletKey", uint(1)).Return("", nil)
|
||||
|
|
|
|||
|
|
@ -23,7 +23,7 @@ import (
|
|||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
func (svc *nip47Service) HandleEvent(ctx context.Context, relay nostrmodels.Relay, event *nostr.Event, lnClient lnclient.LNClient) {
|
||||
func (svc *nip47Service) HandleEvent(ctx context.Context, pool nostrmodels.SimplePool, event *nostr.Event, lnClient lnclient.LNClient) {
|
||||
var nip47Response *models.Response
|
||||
logger.Logger.WithFields(logrus.Fields{
|
||||
"requestEventNostrId": event.ID,
|
||||
|
|
@ -155,7 +155,7 @@ func (svc *nip47Service) HandleEvent(ctx context.Context, relay nostrmodels.Rela
|
|||
"eventKind": event.Kind,
|
||||
}).WithError(err).Error("Failed to process event")
|
||||
}
|
||||
svc.publishResponseEvent(ctx, relay, &requestEvent, resp, &app)
|
||||
svc.publishResponseEvent(ctx, pool, &requestEvent, resp, &app)
|
||||
|
||||
return
|
||||
}
|
||||
|
|
@ -182,7 +182,7 @@ func (svc *nip47Service) HandleEvent(ctx context.Context, relay nostrmodels.Rela
|
|||
"eventKind": event.Kind,
|
||||
}).WithError(err).Error("Failed to process event")
|
||||
}
|
||||
svc.publishResponseEvent(ctx, relay, &requestEvent, resp, &app)
|
||||
svc.publishResponseEvent(ctx, pool, &requestEvent, resp, &app)
|
||||
|
||||
err = svc.db.
|
||||
Model(&requestEvent).
|
||||
|
|
@ -244,7 +244,7 @@ func (svc *nip47Service) HandleEvent(ctx context.Context, relay nostrmodels.Rela
|
|||
"eventKind": event.Kind,
|
||||
}).WithError(err).Error("Failed to process event")
|
||||
}
|
||||
svc.publishResponseEvent(ctx, relay, &requestEvent, resp, &app)
|
||||
svc.publishResponseEvent(ctx, pool, &requestEvent, resp, &app)
|
||||
|
||||
return
|
||||
}
|
||||
|
|
@ -287,7 +287,7 @@ func (svc *nip47Service) HandleEvent(ctx context.Context, relay nostrmodels.Rela
|
|||
}).WithError(err).Error("Failed to create response")
|
||||
state = db.REQUEST_EVENT_STATE_HANDLER_ERROR
|
||||
} else {
|
||||
err = svc.publishResponseEvent(ctx, relay, &requestEvent, resp, &app)
|
||||
err = svc.publishResponseEvent(ctx, pool, &requestEvent, resp, &app)
|
||||
if err != nil {
|
||||
logger.Logger.WithFields(logrus.Fields{
|
||||
"requestEventNostrId": event.ID,
|
||||
|
|
@ -476,7 +476,7 @@ func (svc *nip47Service) CreateResponse(initialEvent *nostr.Event, content inter
|
|||
return resp, nil
|
||||
}
|
||||
|
||||
func (svc *nip47Service) publishResponseEvent(ctx context.Context, relay nostrmodels.Relay, requestEvent *db.RequestEvent, resp *nostr.Event, app *db.App) error {
|
||||
func (svc *nip47Service) publishResponseEvent(ctx context.Context, pool nostrmodels.SimplePool, requestEvent *db.RequestEvent, resp *nostr.Event, app *db.App) error {
|
||||
var appId *uint
|
||||
if app != nil {
|
||||
appId = &app.ID
|
||||
|
|
@ -493,8 +493,25 @@ func (svc *nip47Service) publishResponseEvent(ctx context.Context, relay nostrmo
|
|||
}
|
||||
|
||||
updateColumns := make(map[string]interface{})
|
||||
err = relay.Publish(ctx, *resp)
|
||||
if err != nil {
|
||||
publishResultChannel := pool.PublishMany(ctx, svc.cfg.GetRelayUrls(), *resp)
|
||||
|
||||
publishSuccessful := false
|
||||
for result := range publishResultChannel {
|
||||
if result.Error == nil {
|
||||
publishSuccessful = true
|
||||
} else {
|
||||
logger.Logger.WithFields(logrus.Fields{
|
||||
"requestEventId": requestEvent.ID,
|
||||
"requestNostrEventId": requestEvent.NostrId,
|
||||
"appId": appId,
|
||||
"responseEventId": responseEvent.ID,
|
||||
"responseNostrEventId": resp.ID,
|
||||
"relay": result.RelayURL,
|
||||
}).WithError(result.Error).Error("failed to publish response event to relay")
|
||||
}
|
||||
}
|
||||
|
||||
if !publishSuccessful {
|
||||
updateColumns["state"] = db.RESPONSE_EVENT_STATE_PUBLISH_FAILED
|
||||
logger.Logger.WithFields(logrus.Fields{
|
||||
"requestEventId": requestEvent.ID,
|
||||
|
|
|
|||
|
|
@ -20,8 +20,6 @@ import (
|
|||
"github.com/getAlby/hub/tests"
|
||||
)
|
||||
|
||||
// TODO: test HandleEvent
|
||||
// TODO: test a request cannot be processed twice
|
||||
// TODO: test if an app doesn't exist it returns the right error code
|
||||
|
||||
func TestCreateResponse_Nip04(t *testing.T) {
|
||||
|
|
@ -149,14 +147,14 @@ func doTestHandleResponse_WithPermission(t *testing.T, svc *tests.TestService, c
|
|||
err = reqEvent.Sign(reqPrivateKey)
|
||||
assert.NoError(t, err)
|
||||
|
||||
relay := tests.NewMockRelay()
|
||||
pool := tests.NewMockSimplePool()
|
||||
|
||||
nip47svc.HandleEvent(context.TODO(), relay, reqEvent, svc.LNClient)
|
||||
nip47svc.HandleEvent(context.TODO(), pool, reqEvent, svc.LNClient)
|
||||
|
||||
assert.NotNil(t, relay.PublishedEvents[0])
|
||||
assert.NotEmpty(t, relay.PublishedEvents[0].Content)
|
||||
assert.NotNil(t, pool.PublishedEvents[0])
|
||||
assert.NotEmpty(t, pool.PublishedEvents[0].Content)
|
||||
|
||||
decrypted, err := cipher.Decrypt(relay.PublishedEvents[0].Content)
|
||||
decrypted, err := cipher.Decrypt(pool.PublishedEvents[0].Content)
|
||||
assert.NoError(t, err)
|
||||
|
||||
type getInfoResult struct {
|
||||
|
|
@ -238,19 +236,19 @@ func doTestHandleResponse_DuplicateRequest(t *testing.T, svc *tests.TestService,
|
|||
err = reqEvent.Sign(reqPrivateKey)
|
||||
assert.NoError(t, err)
|
||||
|
||||
relay := tests.NewMockRelay()
|
||||
pool := tests.NewMockSimplePool()
|
||||
|
||||
nip47svc.HandleEvent(context.TODO(), relay, reqEvent, svc.LNClient)
|
||||
nip47svc.HandleEvent(context.TODO(), pool, reqEvent, svc.LNClient)
|
||||
|
||||
assert.NotNil(t, relay.PublishedEvents[0])
|
||||
assert.NotEmpty(t, relay.PublishedEvents[0].Content)
|
||||
assert.NotNil(t, pool.PublishedEvents[0])
|
||||
assert.NotEmpty(t, pool.PublishedEvents[0].Content)
|
||||
|
||||
relay.PublishedEvents = nil
|
||||
pool.PublishedEvents = nil
|
||||
|
||||
nip47svc.HandleEvent(context.TODO(), relay, reqEvent, svc.LNClient)
|
||||
nip47svc.HandleEvent(context.TODO(), pool, reqEvent, svc.LNClient)
|
||||
|
||||
// second time it should not publish
|
||||
assert.Nil(t, relay.PublishedEvents)
|
||||
assert.Nil(t, pool.PublishedEvents)
|
||||
}
|
||||
|
||||
func TestHandleResponse_Nip04_NoPermission(t *testing.T) {
|
||||
|
|
@ -305,14 +303,14 @@ func doTestHandleResponse_NoPermission(t *testing.T, svc *tests.TestService, cre
|
|||
err = reqEvent.Sign(reqPrivateKey)
|
||||
assert.NoError(t, err)
|
||||
|
||||
relay := tests.NewMockRelay()
|
||||
pool := tests.NewMockSimplePool()
|
||||
|
||||
nip47svc.HandleEvent(context.TODO(), relay, reqEvent, svc.LNClient)
|
||||
nip47svc.HandleEvent(context.TODO(), pool, reqEvent, svc.LNClient)
|
||||
|
||||
assert.NotNil(t, relay.PublishedEvents[0])
|
||||
assert.NotEmpty(t, relay.PublishedEvents[0].Content)
|
||||
assert.NotNil(t, pool.PublishedEvents[0])
|
||||
assert.NotEmpty(t, pool.PublishedEvents[0].Content)
|
||||
|
||||
decrypted, err := cipher.Decrypt(relay.PublishedEvents[0].Content)
|
||||
decrypted, err := cipher.Decrypt(pool.PublishedEvents[0].Content)
|
||||
assert.NoError(t, err)
|
||||
|
||||
unmarshalledResponse := models.Response{}
|
||||
|
|
@ -385,20 +383,20 @@ func doTestHandleResponse_OldRequestForPayment(t *testing.T, svc *tests.TestServ
|
|||
err = reqEvent.Sign(reqPrivateKey)
|
||||
assert.NoError(t, err)
|
||||
|
||||
relay := tests.NewMockRelay()
|
||||
pool := tests.NewMockSimplePool()
|
||||
|
||||
nip47svc.HandleEvent(context.TODO(), relay, reqEvent, svc.LNClient)
|
||||
nip47svc.HandleEvent(context.TODO(), pool, reqEvent, svc.LNClient)
|
||||
|
||||
// it shouldn't return anything for an old request
|
||||
assert.Nil(t, relay.PublishedEvents)
|
||||
assert.Nil(t, pool.PublishedEvents)
|
||||
|
||||
// change the request to now
|
||||
reqEvent.CreatedAt = nostr.Now()
|
||||
err = reqEvent.Sign(reqPrivateKey)
|
||||
assert.NoError(t, err)
|
||||
|
||||
nip47svc.HandleEvent(context.TODO(), relay, reqEvent, svc.LNClient)
|
||||
assert.NotNil(t, relay.PublishedEvents)
|
||||
nip47svc.HandleEvent(context.TODO(), pool, reqEvent, svc.LNClient)
|
||||
assert.NotNil(t, pool.PublishedEvents)
|
||||
}
|
||||
|
||||
func TestHandleResponse_Nip04_IncorrectPubkey(t *testing.T) {
|
||||
|
|
@ -465,11 +463,11 @@ func doTestHandleResponse_IncorrectPubkey(t *testing.T, svc *tests.TestService,
|
|||
// set a different pubkey (this will not pass validation)
|
||||
reqEvent.PubKey = reqPubkey
|
||||
|
||||
relay := tests.NewMockRelay()
|
||||
pool := tests.NewMockSimplePool()
|
||||
|
||||
nip47svc.HandleEvent(context.TODO(), relay, reqEvent, svc.LNClient)
|
||||
nip47svc.HandleEvent(context.TODO(), pool, reqEvent, svc.LNClient)
|
||||
|
||||
assert.Nil(t, relay.PublishedEvents)
|
||||
assert.Nil(t, pool.PublishedEvents)
|
||||
}
|
||||
|
||||
func TestHandleResponse_NoApp(t *testing.T) {
|
||||
|
|
@ -510,12 +508,12 @@ func TestHandleResponse_NoApp(t *testing.T) {
|
|||
err = reqEvent.Sign(reqPrivateKey)
|
||||
assert.NoError(t, err)
|
||||
|
||||
relay := tests.NewMockRelay()
|
||||
pool := tests.NewMockSimplePool()
|
||||
|
||||
nip47svc.HandleEvent(context.TODO(), relay, reqEvent, svc.LNClient)
|
||||
nip47svc.HandleEvent(context.TODO(), pool, reqEvent, svc.LNClient)
|
||||
|
||||
// it shouldn't return anything for an invalid app key
|
||||
assert.Nil(t, relay.PublishedEvents)
|
||||
assert.Nil(t, pool.PublishedEvents)
|
||||
}
|
||||
|
||||
func TestHandleResponse_UnknownEncryptionTag(t *testing.T) {
|
||||
|
|
@ -569,12 +567,12 @@ func doTestHandleResponse_UnknownEncryptionTag(t *testing.T, svc *tests.TestServ
|
|||
err = reqEvent.Sign(reqPrivateKey)
|
||||
assert.NoError(t, err)
|
||||
|
||||
relay := tests.NewMockRelay()
|
||||
pool := tests.NewMockSimplePool()
|
||||
|
||||
nip47svc.HandleEvent(context.TODO(), relay, reqEvent, svc.LNClient)
|
||||
nip47svc.HandleEvent(context.TODO(), pool, reqEvent, svc.LNClient)
|
||||
|
||||
assert.NotNil(t, relay.PublishedEvents)
|
||||
responseContent := relay.PublishedEvents[0].Content
|
||||
assert.NotNil(t, pool.PublishedEvents)
|
||||
responseContent := pool.PublishedEvents[0].Content
|
||||
msg, err = cipher.Decrypt(responseContent)
|
||||
assert.NoError(t, err)
|
||||
assert.NotEqual(t, "", msg)
|
||||
|
|
@ -650,12 +648,12 @@ func doTestHandleResponse_EncryptionTagDoesNotMatchPayload(t *testing.T, svc *te
|
|||
err = reqEvent.Sign(reqPrivateKey)
|
||||
assert.NoError(t, err)
|
||||
|
||||
relay := tests.NewMockRelay()
|
||||
pool := tests.NewMockSimplePool()
|
||||
|
||||
nip47svc.HandleEvent(context.TODO(), relay, reqEvent, svc.LNClient)
|
||||
nip47svc.HandleEvent(context.TODO(), pool, reqEvent, svc.LNClient)
|
||||
|
||||
assert.NotNil(t, relay.PublishedEvents)
|
||||
responseContent := relay.PublishedEvents[0].Content
|
||||
assert.NotNil(t, pool.PublishedEvents)
|
||||
responseContent := pool.PublishedEvents[0].Content
|
||||
msg, err = nip44Cipher.Decrypt(responseContent)
|
||||
assert.NoError(t, err)
|
||||
assert.NotEqual(t, "", msg)
|
||||
|
|
|
|||
|
|
@ -17,6 +17,7 @@ import (
|
|||
"github.com/getAlby/hub/service/keys"
|
||||
"github.com/getAlby/hub/transactions"
|
||||
"github.com/nbd-wtf/go-nostr"
|
||||
"github.com/sirupsen/logrus"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
|
|
@ -35,14 +36,14 @@ type nip47Service struct {
|
|||
|
||||
type Nip47Service interface {
|
||||
events.EventSubscriber
|
||||
StartNotifier(relay *nostr.Relay)
|
||||
StartNip47InfoPublisher(relay *nostr.Relay, lnClient lnclient.LNClient)
|
||||
HandleEvent(ctx context.Context, relay nostrmodels.Relay, event *nostr.Event, lnClient lnclient.LNClient)
|
||||
GetNip47Info(ctx context.Context, relay *nostr.Relay, appWalletPubKey string) (*nostr.Event, error)
|
||||
PublishNip47Info(ctx context.Context, relay nostrmodels.Relay, appId uint, appWalletPubKey string, appWalletPrivKey string, lnClient lnclient.LNClient) (*nostr.Event, error)
|
||||
PublishNip47InfoDeletion(ctx context.Context, relay nostrmodels.Relay, appWalletPubKey string, appWalletPrivKey string, infoEventId string) error
|
||||
StartNotifier(ctx context.Context, pool *nostr.SimplePool)
|
||||
StartNip47InfoPublisher(ctx context.Context, pool *nostr.SimplePool, lnClient lnclient.LNClient)
|
||||
HandleEvent(ctx context.Context, pool nostrmodels.SimplePool, event *nostr.Event, lnClient lnclient.LNClient)
|
||||
GetNip47Info(ctx context.Context, pool nostrmodels.SimplePool, appWalletPubKey string) (*nostr.Event, error)
|
||||
PublishNip47Info(ctx context.Context, pool nostrmodels.SimplePool, appId uint, appWalletPubKey string, appWalletPrivKey string, relayUrl string, lnClient lnclient.LNClient) (*nostr.Event, error)
|
||||
PublishNip47InfoDeletion(ctx context.Context, pool nostrmodels.SimplePool, appWalletPubKey string, appWalletPrivKey string, infoEventId string) error
|
||||
CreateResponse(initialEvent *nostr.Event, content interface{}, tags nostr.Tags, cipher *cipher.Nip47Cipher, walletPrivKey string) (result *nostr.Event, err error)
|
||||
EnqueueNip47InfoPublishRequest(appId uint, appWalletPubKey, appWalletPrivKey string)
|
||||
EnqueueNip47InfoPublishRequest(appId uint, appWalletPubKey, appWalletPrivKey, relayUrl string)
|
||||
}
|
||||
|
||||
func NewNip47Service(db *gorm.DB, cfg config.Config, keys keys.Keys, eventPublisher events.EventPublisher, albyOAuthSvc alby.AlbyOAuthService) *nip47Service {
|
||||
|
|
@ -67,17 +68,17 @@ func (svc *nip47Service) ConsumeEvent(ctx context.Context, event *events.Event,
|
|||
// The notifier is decoupled from the notification queue
|
||||
// so that if Alby Hub disconnects from the relay, it will wait to reconnect
|
||||
// to send notifications rather than dropping them
|
||||
func (svc *nip47Service) StartNotifier(relay *nostr.Relay) {
|
||||
nip47Notifier := notifications.NewNip47Notifier(relay, svc.db, svc.cfg, svc.keys, svc.permissionsService)
|
||||
func (svc *nip47Service) StartNotifier(ctx context.Context, pool *nostr.SimplePool) {
|
||||
nip47Notifier := notifications.NewNip47Notifier(pool, svc.db, svc.cfg, svc.keys, svc.permissionsService)
|
||||
go func() {
|
||||
for {
|
||||
select {
|
||||
case <-relay.Context().Done():
|
||||
// relay disconnected
|
||||
case <-ctx.Done():
|
||||
// app exited
|
||||
return
|
||||
case event := <-svc.nip47NotificationQueue.Channel():
|
||||
logger.Logger.WithField("event", event).Debug("Consuming event from notification queue")
|
||||
err := nip47Notifier.ConsumeEvent(relay.Context(), event)
|
||||
err := nip47Notifier.ConsumeEvent(ctx, event)
|
||||
if err != nil {
|
||||
logger.Logger.WithError(err).WithField("event", event).Error("Failed to consume event from notification queue")
|
||||
// wait and then re-add the item to the queue
|
||||
|
|
@ -89,28 +90,42 @@ func (svc *nip47Service) StartNotifier(relay *nostr.Relay) {
|
|||
}()
|
||||
}
|
||||
|
||||
func (svc *nip47Service) EnqueueNip47InfoPublishRequest(appId uint, appWalletPubKey, appWalletPrivKey string) {
|
||||
func (svc *nip47Service) EnqueueNip47InfoPublishRequest(appId uint, appWalletPubKey, appWalletPrivKey, relayUrl string) {
|
||||
svc.enqueueNip47InfoPublishRequestWithAttempt(appId, appWalletPubKey, appWalletPrivKey, relayUrl, 0)
|
||||
}
|
||||
|
||||
func (svc *nip47Service) enqueueNip47InfoPublishRequestWithAttempt(appId uint, appWalletPubKey, appWalletPrivKey, relayUrl string, attempt uint32) {
|
||||
svc.nip47InfoPublishQueue.AddToQueue(&Nip47InfoPublishRequest{
|
||||
AppId: appId,
|
||||
AppWalletPubKey: appWalletPubKey,
|
||||
AppWalletPrivKey: appWalletPrivKey,
|
||||
RelayUrl: relayUrl,
|
||||
Attempt: attempt,
|
||||
})
|
||||
}
|
||||
|
||||
func (svc *nip47Service) StartNip47InfoPublisher(relay *nostr.Relay, lnClient lnclient.LNClient) {
|
||||
func (svc *nip47Service) StartNip47InfoPublisher(ctx context.Context, pool *nostr.SimplePool, lnClient lnclient.LNClient) {
|
||||
go func() {
|
||||
for {
|
||||
select {
|
||||
case <-relay.Context().Done():
|
||||
case <-ctx.Done():
|
||||
// relay disconnected
|
||||
return
|
||||
case req := <-svc.nip47InfoPublishQueue.Channel():
|
||||
_, err := svc.PublishNip47Info(relay.Context(), relay, req.AppId, req.AppWalletPubKey, req.AppWalletPrivKey, lnClient)
|
||||
_, err := svc.PublishNip47Info(ctx, pool, req.AppId, req.AppWalletPubKey, req.AppWalletPrivKey, req.RelayUrl, lnClient)
|
||||
if err != nil {
|
||||
logger.Logger.WithError(err).WithField("wallet_pubkey", req.AppWalletPubKey).Error("Failed to publish NIP47 info from queue")
|
||||
logger.Logger.WithError(err).WithFields(logrus.Fields{
|
||||
"wallet_pubkey": req.AppWalletPubKey,
|
||||
"relay_url": req.RelayUrl,
|
||||
}).Error("Failed to publish NIP47 info from queue")
|
||||
|
||||
// wait and then re-add the item to the queue
|
||||
time.Sleep(5 * time.Second)
|
||||
svc.EnqueueNip47InfoPublishRequest(req.AppId, req.AppWalletPubKey, req.AppWalletPrivKey)
|
||||
// done async to ensure an offline relay does not delay
|
||||
// the publishing of newly created app connections
|
||||
go func() {
|
||||
time.Sleep((5 * time.Duration(req.Attempt+1)) * time.Second)
|
||||
svc.enqueueNip47InfoPublishRequestWithAttempt(req.AppId, req.AppWalletPubKey, req.AppWalletPrivKey, req.RelayUrl, req.Attempt+1)
|
||||
}()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -21,16 +21,16 @@ import (
|
|||
)
|
||||
|
||||
type Nip47Notifier struct {
|
||||
relay nostrmodels.Relay
|
||||
pool nostrmodels.SimplePool
|
||||
cfg config.Config
|
||||
keys keys.Keys
|
||||
db *gorm.DB
|
||||
permissionsSvc permissions.PermissionsService
|
||||
}
|
||||
|
||||
func NewNip47Notifier(relay nostrmodels.Relay, db *gorm.DB, cfg config.Config, keys keys.Keys, permissionsSvc permissions.PermissionsService) *Nip47Notifier {
|
||||
func NewNip47Notifier(pool nostrmodels.SimplePool, db *gorm.DB, cfg config.Config, keys keys.Keys, permissionsSvc permissions.PermissionsService) *Nip47Notifier {
|
||||
return &Nip47Notifier{
|
||||
relay: relay,
|
||||
pool: pool,
|
||||
cfg: cfg,
|
||||
db: db,
|
||||
permissionsSvc: permissionsSvc,
|
||||
|
|
@ -212,8 +212,22 @@ func (notifier *Nip47Notifier) notifySubscriber(ctx context.Context, app *db.App
|
|||
return err
|
||||
}
|
||||
|
||||
err = notifier.relay.Publish(ctx, *event)
|
||||
if err != nil {
|
||||
publishResultChannel := notifier.pool.PublishMany(ctx, notifier.cfg.GetRelayUrls(), *event)
|
||||
|
||||
publishSuccessful := false
|
||||
for result := range publishResultChannel {
|
||||
if result.Error == nil {
|
||||
publishSuccessful = true
|
||||
} else {
|
||||
logger.Logger.WithFields(logrus.Fields{
|
||||
"notification": notification,
|
||||
"appId": app.ID,
|
||||
"relay": result.RelayURL,
|
||||
}).WithError(result.Error).Error("failed to publish notification to relay")
|
||||
}
|
||||
}
|
||||
|
||||
if !publishSuccessful {
|
||||
logger.Logger.WithFields(logrus.Fields{
|
||||
"notification": notification,
|
||||
"appId": app.ID,
|
||||
|
|
|
|||
|
|
@ -76,18 +76,18 @@ func doTestSendNotificationPaymentReceived(t *testing.T, svc *tests.TestService,
|
|||
receivedEvent := <-nip47NotificationQueue.Channel()
|
||||
assert.Equal(t, testEvent, receivedEvent)
|
||||
|
||||
relay := tests.NewMockRelay()
|
||||
pool := tests.NewMockSimplePool()
|
||||
|
||||
permissionsSvc := permissions.NewPermissionsService(svc.DB, svc.EventPublisher)
|
||||
|
||||
notifier := NewNip47Notifier(relay, svc.DB, svc.Cfg, svc.Keys, permissionsSvc)
|
||||
notifier := NewNip47Notifier(pool, svc.DB, svc.Cfg, svc.Keys, permissionsSvc)
|
||||
notifier.ConsumeEvent(ctx, receivedEvent)
|
||||
|
||||
var publishedEvent *nostr.Event
|
||||
if nip47Encryption == constants.ENCRYPTION_TYPE_NIP04 {
|
||||
publishedEvent = relay.PublishedEvents[0]
|
||||
publishedEvent = pool.PublishedEvents[0]
|
||||
} else {
|
||||
publishedEvent = relay.PublishedEvents[1]
|
||||
publishedEvent = pool.PublishedEvents[1]
|
||||
}
|
||||
|
||||
assert.NotNil(t, publishedEvent)
|
||||
|
|
@ -190,18 +190,18 @@ func doTestSendNotificationPaymentSent(t *testing.T, svc *tests.TestService, cre
|
|||
receivedEvent := <-nip47NotificationQueue.Channel()
|
||||
assert.Equal(t, testEvent, receivedEvent)
|
||||
|
||||
relay := tests.NewMockRelay()
|
||||
pool := tests.NewMockSimplePool()
|
||||
|
||||
permissionsSvc := permissions.NewPermissionsService(svc.DB, svc.EventPublisher)
|
||||
|
||||
notifier := NewNip47Notifier(relay, svc.DB, svc.Cfg, svc.Keys, permissionsSvc)
|
||||
notifier := NewNip47Notifier(pool, svc.DB, svc.Cfg, svc.Keys, permissionsSvc)
|
||||
notifier.ConsumeEvent(ctx, receivedEvent)
|
||||
|
||||
var publishedEvent *nostr.Event
|
||||
if nip47Encryption == constants.ENCRYPTION_TYPE_NIP04 {
|
||||
publishedEvent = relay.PublishedEvents[0]
|
||||
publishedEvent = pool.PublishedEvents[0]
|
||||
} else {
|
||||
publishedEvent = relay.PublishedEvents[1]
|
||||
publishedEvent = pool.PublishedEvents[1]
|
||||
}
|
||||
|
||||
assert.NotNil(t, publishedEvent)
|
||||
|
|
@ -283,14 +283,14 @@ func doTestSendNotificationNoPermission(t *testing.T, svc *tests.TestService) {
|
|||
receivedEvent := <-nip47NotificationQueue.Channel()
|
||||
assert.Equal(t, testEvent, receivedEvent)
|
||||
|
||||
relay := tests.NewMockRelay()
|
||||
pool := tests.NewMockSimplePool()
|
||||
|
||||
permissionsSvc := permissions.NewPermissionsService(svc.DB, svc.EventPublisher)
|
||||
|
||||
notifier := NewNip47Notifier(relay, svc.DB, svc.Cfg, svc.Keys, permissionsSvc)
|
||||
notifier := NewNip47Notifier(pool, svc.DB, svc.Cfg, svc.Keys, permissionsSvc)
|
||||
notifier.ConsumeEvent(ctx, receivedEvent)
|
||||
|
||||
assert.Nil(t, relay.PublishedEvents)
|
||||
assert.Nil(t, pool.PublishedEvents)
|
||||
}
|
||||
|
||||
func TestSendNotification_NoPermission(t *testing.T) {
|
||||
|
|
|
|||
|
|
@ -92,12 +92,12 @@ func (svc *permissionsService) GetPermittedMethods(app *db.App, lnClient lnclien
|
|||
|
||||
func (svc *permissionsService) PermitsNotifications(app *db.App) bool {
|
||||
notificationPermission := db.AppPermission{}
|
||||
err := svc.db.First(¬ificationPermission, &db.AppPermission{
|
||||
result := svc.db.Limit(1).Find(¬ificationPermission, &db.AppPermission{
|
||||
AppId: app.ID,
|
||||
Scope: constants.NOTIFICATIONS_SCOPE,
|
||||
}).Error
|
||||
})
|
||||
|
||||
return err == nil
|
||||
return result.Error == nil && result.RowsAffected > 0
|
||||
}
|
||||
|
||||
func scopesToRequestMethods(scopes []string) []string {
|
||||
|
|
|
|||
|
|
@ -20,6 +20,8 @@ type Nip47InfoPublishRequest struct {
|
|||
AppId uint
|
||||
AppWalletPubKey string
|
||||
AppWalletPrivKey string
|
||||
RelayUrl string
|
||||
Attempt uint32
|
||||
}
|
||||
|
||||
type nip47InfoPublishQueue struct {
|
||||
|
|
@ -43,26 +45,22 @@ func (q *nip47InfoPublishQueue) Channel() <-chan *Nip47InfoPublishRequest {
|
|||
return q.channel
|
||||
}
|
||||
|
||||
func (svc *nip47Service) GetNip47Info(ctx context.Context, relay *nostr.Relay, appWalletPubKey string) (*nostr.Event, error) {
|
||||
func (svc *nip47Service) GetNip47Info(ctx context.Context, pool nostrmodels.SimplePool, appWalletPubKey string) (*nostr.Event, error) {
|
||||
filter := nostr.Filter{
|
||||
Kinds: []int{models.INFO_EVENT_KIND},
|
||||
Authors: []string{appWalletPubKey},
|
||||
Limit: 1,
|
||||
}
|
||||
|
||||
events, err := relay.QuerySync(ctx, filter)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if len(events) == 0 {
|
||||
relayEvent := pool.QuerySingle(ctx, svc.cfg.GetRelayUrls(), filter)
|
||||
if relayEvent == nil {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
return events[0], nil
|
||||
return relayEvent.Event, nil
|
||||
}
|
||||
|
||||
func (svc *nip47Service) PublishNip47Info(ctx context.Context, relay nostrmodels.Relay, appId uint, appWalletPubKey string, appWalletPrivKey string, lnClient lnclient.LNClient) (*nostr.Event, error) {
|
||||
func (svc *nip47Service) PublishNip47Info(ctx context.Context, pool nostrmodels.SimplePool, appId uint, appWalletPubKey string, appWalletPrivKey string, relayUrl string, lnClient lnclient.LNClient) (*nostr.Event, error) {
|
||||
var capabilities []string
|
||||
var permitsNotifications bool
|
||||
tags := nostr.Tags{[]string{"encryption", cipher.SUPPORTED_ENCRYPTIONS}}
|
||||
|
|
@ -103,15 +101,29 @@ func (svc *nip47Service) PublishNip47Info(ctx context.Context, relay nostrmodels
|
|||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
err = relay.Publish(ctx, *ev)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("nostr publish not successful: %s", err)
|
||||
|
||||
// publish to a single relay so that we can requeue failed publishes on a relay level
|
||||
publishResultChannel := pool.PublishMany(ctx, []string{relayUrl}, *ev)
|
||||
|
||||
publishSuccessful := false
|
||||
for result := range publishResultChannel {
|
||||
if result.Error == nil {
|
||||
publishSuccessful = true
|
||||
} else {
|
||||
logger.Logger.WithFields(logrus.Fields{
|
||||
"appId": appId,
|
||||
"relay": result.RelayURL,
|
||||
}).WithError(result.Error).Error("failed to publish nip47 info to relay")
|
||||
}
|
||||
}
|
||||
if !publishSuccessful {
|
||||
return nil, fmt.Errorf("nostr publish failed: %s", err)
|
||||
}
|
||||
logger.Logger.WithField("wallet_pubkey", appWalletPubKey).Debug("published info event")
|
||||
return ev, nil
|
||||
}
|
||||
|
||||
func (svc *nip47Service) PublishNip47InfoDeletion(ctx context.Context, relay nostrmodels.Relay, appWalletPubKey string, appWalletPrivKey string, infoEventId string) error {
|
||||
func (svc *nip47Service) PublishNip47InfoDeletion(ctx context.Context, pool nostrmodels.SimplePool, appWalletPubKey string, appWalletPrivKey string, infoEventId string) error {
|
||||
ev := &nostr.Event{}
|
||||
ev.Kind = nostr.KindDeletion
|
||||
ev.Content = "deleting nip47 info since app connection for this key was deleted"
|
||||
|
|
@ -122,9 +134,22 @@ func (svc *nip47Service) PublishNip47InfoDeletion(ctx context.Context, relay nos
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = relay.Publish(ctx, *ev)
|
||||
if err != nil {
|
||||
return fmt.Errorf("nostr publish not successful: %s", err)
|
||||
publishResultChannel := pool.PublishMany(ctx, svc.cfg.GetRelayUrls(), *ev)
|
||||
|
||||
publishSuccessful := false
|
||||
for result := range publishResultChannel {
|
||||
if result.Error == nil {
|
||||
publishSuccessful = true
|
||||
} else {
|
||||
logger.Logger.WithFields(logrus.Fields{
|
||||
"wallet_pubkey": appWalletPubKey,
|
||||
"relay": result.RelayURL,
|
||||
}).WithError(result.Error).Error("failed to publish info event deletion to relay")
|
||||
}
|
||||
}
|
||||
|
||||
if !publishSuccessful {
|
||||
return fmt.Errorf("failed to publish info event deletion to all relays")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -6,6 +6,12 @@ import (
|
|||
"github.com/nbd-wtf/go-nostr"
|
||||
)
|
||||
|
||||
type Relay interface {
|
||||
Publish(ctx context.Context, event nostr.Event) error
|
||||
type SimplePool interface {
|
||||
PublishMany(ctx context.Context, relayUrls []string, event nostr.Event) chan nostr.PublishResult
|
||||
QuerySingle(
|
||||
ctx context.Context,
|
||||
urls []string,
|
||||
filter nostr.Filter,
|
||||
opts ...nostr.SubscriptionOption,
|
||||
) *nostr.RelayEvent
|
||||
}
|
||||
|
|
|
|||
|
|
@ -14,8 +14,8 @@ import (
|
|||
|
||||
type createAppConsumer struct {
|
||||
events.EventSubscriber
|
||||
svc *service
|
||||
relay *nostr.Relay
|
||||
svc *service
|
||||
pool *nostr.SimplePool
|
||||
}
|
||||
|
||||
// When a new app is created, subscribe to it on the relay
|
||||
|
|
@ -56,10 +56,12 @@ func (s *createAppConsumer) ConsumeEvent(ctx context.Context, event *events.Even
|
|||
logger.Logger.WithError(err).Error("Failed to calculate app wallet pub key")
|
||||
return
|
||||
}
|
||||
s.svc.nip47Service.EnqueueNip47InfoPublishRequest(id, walletPubKey, walletPrivKey)
|
||||
for _, relayUrl := range s.svc.cfg.GetRelayUrls() {
|
||||
s.svc.nip47Service.EnqueueNip47InfoPublishRequest(id, walletPubKey, walletPrivKey, relayUrl)
|
||||
}
|
||||
|
||||
go func() {
|
||||
err = s.svc.startAppWalletSubscription(ctx, s.relay, walletPubKey)
|
||||
err = s.svc.startAppWalletSubscription(ctx, s.pool, walletPubKey)
|
||||
if err != nil && !errors.Is(err, context.Canceled) {
|
||||
logger.Logger.WithError(err).WithFields(logrus.Fields{
|
||||
"app_id": id}).Error("Failed to subscribe to wallet")
|
||||
|
|
|
|||
|
|
@ -10,10 +10,10 @@ import (
|
|||
|
||||
type deleteAppConsumer struct {
|
||||
events.EventSubscriber
|
||||
walletPubkey string
|
||||
relay *nostr.Relay
|
||||
nostrSubscription *nostr.Subscription
|
||||
svc *service
|
||||
walletPubkey string
|
||||
pool *nostr.SimplePool
|
||||
cancelSubscription func()
|
||||
svc *service
|
||||
}
|
||||
|
||||
// When an app is deleted, unsubscribe from events for that app on the relay
|
||||
|
|
@ -46,19 +46,21 @@ func (s *deleteAppConsumer) ConsumeEvent(ctx context.Context, event *events.Even
|
|||
// Note: for legacy apps this check will always return false as the wallet pubkey
|
||||
// generated by the id will not match the master key which is used for all legacy apps
|
||||
if s.walletPubkey == walletPubKey {
|
||||
s.nostrSubscription.Unsub()
|
||||
// no longer need to listen to events for this wallet
|
||||
s.cancelSubscription()
|
||||
|
||||
// remove this consumer as subscriber in eventPublisher
|
||||
s.svc.eventPublisher.RemoveSubscriber(s)
|
||||
|
||||
// try to delete info event from relays (non-critical if it fails)
|
||||
// get nip47 event info for this app wallet key
|
||||
nip47InfoEvent, err := s.svc.GetNip47Service().GetNip47Info(ctx, s.relay, s.walletPubkey)
|
||||
nip47InfoEvent, err := s.svc.GetNip47Service().GetNip47Info(ctx, s.pool, s.walletPubkey)
|
||||
if err != nil {
|
||||
logger.Logger.WithError(err).Error("Could not get nip47 info event")
|
||||
return
|
||||
}
|
||||
if nip47InfoEvent != nil {
|
||||
err = s.svc.nip47Service.PublishNip47InfoDeletion(ctx, s.relay, walletPubKey, walletPrivKey, nip47InfoEvent.ID)
|
||||
err = s.svc.nip47Service.PublishNip47InfoDeletion(ctx, s.pool, walletPubKey, walletPrivKey, nip47InfoEvent.ID)
|
||||
if err != nil {
|
||||
logger.Logger.WithError(err).WithField("event", event).Error("Failed to publish nip47 info deletion")
|
||||
}
|
||||
|
|
|
|||
|
|
@ -12,6 +12,11 @@ import (
|
|||
"github.com/getAlby/hub/transactions"
|
||||
)
|
||||
|
||||
type RelayStatus struct {
|
||||
Url string
|
||||
Online bool
|
||||
}
|
||||
|
||||
type Service interface {
|
||||
StartApp(encryptionKey string) error
|
||||
StopApp()
|
||||
|
|
@ -27,6 +32,6 @@ type Service interface {
|
|||
GetDB() *gorm.DB
|
||||
GetConfig() config.Config
|
||||
GetKeys() keys.Keys
|
||||
IsRelayReady() bool
|
||||
GetRelayStatuses() []RelayStatus
|
||||
GetStartupState() string
|
||||
}
|
||||
|
|
|
|||
|
|
@ -6,11 +6,9 @@ import (
|
|||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/adrg/xdg"
|
||||
"github.com/nbd-wtf/go-nostr"
|
||||
"github.com/sirupsen/logrus"
|
||||
"gorm.io/gorm"
|
||||
|
||||
|
|
@ -29,7 +27,6 @@ import (
|
|||
"github.com/getAlby/hub/db"
|
||||
"github.com/getAlby/hub/lnclient"
|
||||
"github.com/getAlby/hub/nip47"
|
||||
"github.com/getAlby/hub/nip47/models"
|
||||
)
|
||||
|
||||
type service struct {
|
||||
|
|
@ -47,7 +44,7 @@ type service struct {
|
|||
nip47Service nip47.Nip47Service
|
||||
appCancelFn context.CancelFunc
|
||||
keys keys.Keys
|
||||
isRelayReady atomic.Bool
|
||||
relayStatuses []RelayStatus
|
||||
startupState string
|
||||
}
|
||||
|
||||
|
|
@ -177,14 +174,6 @@ func NewService(ctx context.Context) (*service, error) {
|
|||
return svc, nil
|
||||
}
|
||||
|
||||
func (svc *service) createFilters(identityPubkey string) nostr.Filters {
|
||||
filter := nostr.Filter{
|
||||
Tags: nostr.TagMap{"p": []string{identityPubkey}},
|
||||
Kinds: []int{models.REQUEST_KIND},
|
||||
}
|
||||
return []nostr.Filter{filter}
|
||||
}
|
||||
|
||||
func (svc *service) noticeHandler(notice string) {
|
||||
logger.Logger.Infof("Received a notice %s", notice)
|
||||
}
|
||||
|
|
@ -282,12 +271,8 @@ func (svc *service) GetKeys() keys.Keys {
|
|||
return svc.keys
|
||||
}
|
||||
|
||||
func (svc *service) setRelayReady(ready bool) {
|
||||
svc.isRelayReady.Store(ready)
|
||||
}
|
||||
|
||||
func (svc *service) IsRelayReady() bool {
|
||||
return svc.isRelayReady.Load()
|
||||
func (svc *service) GetRelayStatuses() []RelayStatus {
|
||||
return svc.relayStatuses
|
||||
}
|
||||
|
||||
func (svc *service) GetStartupState() string {
|
||||
|
|
|
|||
248
service/start.go
248
service/start.go
|
|
@ -10,6 +10,7 @@ import (
|
|||
"time"
|
||||
|
||||
"github.com/getAlby/hub/db"
|
||||
"github.com/getAlby/hub/nip47/models"
|
||||
"github.com/getAlby/hub/swaps"
|
||||
"github.com/getAlby/hub/version"
|
||||
|
||||
|
|
@ -28,7 +29,7 @@ import (
|
|||
)
|
||||
|
||||
func (svc *service) startNostr(ctx context.Context) error {
|
||||
relayUrl := svc.cfg.GetRelayUrl()
|
||||
relayUrls := svc.cfg.GetRelayUrls()
|
||||
|
||||
npub, err := nip19.EncodePublicKey(svc.keys.GetNostrPublicKey())
|
||||
if err != nil {
|
||||
|
|
@ -37,123 +38,83 @@ func (svc *service) startNostr(ctx context.Context) error {
|
|||
}
|
||||
|
||||
logger.Logger.WithFields(logrus.Fields{
|
||||
"npub": npub,
|
||||
"hex": svc.keys.GetNostrPublicKey(),
|
||||
"version": version.Tag,
|
||||
"npub": npub,
|
||||
"hex": svc.keys.GetNostrPublicKey(),
|
||||
"version": version.Tag,
|
||||
"relay_urls": relayUrls,
|
||||
}).Info("Starting Alby Hub")
|
||||
svc.wg.Add(1)
|
||||
|
||||
// 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},
|
||||
}),
|
||||
))
|
||||
|
||||
go func() {
|
||||
// ensure the relay is properly disconnected before exiting
|
||||
defer svc.wg.Done()
|
||||
// Start infinite loop which will be only broken by canceling ctx (SIGINT)
|
||||
var relay *nostr.Relay
|
||||
waitToReconnectSeconds := 0
|
||||
var createAppEventListener events.EventSubscriber
|
||||
var updateAppEventListener events.EventSubscriber
|
||||
for i := 0; ; i++ {
|
||||
// wait for a delay if any before retrying
|
||||
contextCancelled := false
|
||||
|
||||
svc.setRelayReady(false)
|
||||
|
||||
select {
|
||||
case <-ctx.Done(): // application service context cancelled
|
||||
logger.Logger.Info("service context cancelled")
|
||||
contextCancelled = true
|
||||
case <-time.After(time.Duration(waitToReconnectSeconds) * time.Second): // timeout
|
||||
}
|
||||
if contextCancelled {
|
||||
break
|
||||
}
|
||||
|
||||
closeRelay(relay)
|
||||
|
||||
// connect to the relay
|
||||
logger.Logger.WithFields(logrus.Fields{
|
||||
"relay_url": relayUrl,
|
||||
"iteration": i,
|
||||
}).Info("Connecting to the relay")
|
||||
|
||||
relay, err = nostr.RelayConnect(
|
||||
ctx,
|
||||
relayUrl,
|
||||
nostr.WithNoticeHandler(svc.noticeHandler),
|
||||
nostr.WithRequestHeader(http.Header{
|
||||
"User-Agent": {"AlbyHub/" + version.Tag},
|
||||
}))
|
||||
if err != nil {
|
||||
// exponential backoff from 2 - 60 seconds
|
||||
waitToReconnectSeconds = max(waitToReconnectSeconds, 1)
|
||||
waitToReconnectSeconds *= 2
|
||||
waitToReconnectSeconds = min(waitToReconnectSeconds, 60)
|
||||
logger.Logger.WithFields(logrus.Fields{
|
||||
"iteration": i,
|
||||
"retry_seconds": waitToReconnectSeconds,
|
||||
}).WithError(err).Error("Failed to connect to relay")
|
||||
continue
|
||||
}
|
||||
logger.Logger.WithFields(logrus.Fields{
|
||||
"relay_url": relayUrl,
|
||||
}).Info("Connected to the relay")
|
||||
waitToReconnectSeconds = 0
|
||||
|
||||
svc.nip47Service.StartNotifier(relay)
|
||||
svc.nip47Service.StartNip47InfoPublisher(relay, svc.lnClient)
|
||||
|
||||
// register a subscriber for events of "nwc_app_created" which handles creation of nostr subscription for new app
|
||||
if createAppEventListener != nil {
|
||||
svc.eventPublisher.RemoveSubscriber(createAppEventListener)
|
||||
}
|
||||
createAppEventListener = &createAppConsumer{svc: svc, relay: relay}
|
||||
svc.eventPublisher.RegisterSubscriber(createAppEventListener)
|
||||
|
||||
// register a subscriber for events of "nwc_app_updated" which handles re-publishing of nip47 event info
|
||||
if updateAppEventListener != nil {
|
||||
svc.eventPublisher.RemoveSubscriber(updateAppEventListener)
|
||||
}
|
||||
updateAppEventListener = &updateAppConsumer{svc: svc, relay: relay}
|
||||
svc.eventPublisher.RegisterSubscriber(updateAppEventListener)
|
||||
|
||||
// start each app wallet subscription which have a child derived wallet key
|
||||
svc.startAllExistingAppsWalletSubscriptions(ctx, relay)
|
||||
|
||||
// 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")
|
||||
return
|
||||
}
|
||||
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
|
||||
err = svc.startAppWalletSubscription(ctx, relay, svc.keys.GetNostrPublicKey())
|
||||
if err != nil && !errors.Is(err, context.Canceled) {
|
||||
// err being non-nil means that we have an error on the websocket error channel. In this case we just try to reconnect.
|
||||
logger.Logger.WithError(err).Error("Got an error from the relay while listening to legacy subscription.")
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
svc.setRelayReady(true)
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
logger.Logger.Info("Main context cancelled, exiting...")
|
||||
case <-relay.Context().Done():
|
||||
// err being non-nil means that we have an error on the websocket error channel. In this case we just try to reconnect.
|
||||
if relay.ConnectionError != nil {
|
||||
logger.Logger.WithError(relay.ConnectionError).Error("Got an error from the relay, trying to reconnect")
|
||||
} else {
|
||||
logger.Logger.Error("Relay context cancelled, but no connection error...trying to reconnect")
|
||||
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)
|
||||
}
|
||||
}
|
||||
closeRelay(relay)
|
||||
logger.Logger.Info("Relay subroutine ended")
|
||||
}()
|
||||
|
||||
svc.nip47Service.StartNotifier(ctx, pool)
|
||||
svc.nip47Service.StartNip47InfoPublisher(ctx, pool, svc.lnClient)
|
||||
|
||||
// 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
|
||||
err = svc.startAppWalletSubscription(ctx, pool, svc.keys.GetNostrPublicKey())
|
||||
if err != nil && !errors.Is(err, context.Canceled) {
|
||||
// err being non-nil means that we have an error on the websocket error channel. In this case we just try to reconnect.
|
||||
logger.Logger.WithError(err).Error("Got an error from the relay while listening to legacy subscription.")
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
|
|
@ -170,7 +131,9 @@ func (svc *service) publishAllAppInfoEvents() {
|
|||
}
|
||||
if legacyAppCount > 0 {
|
||||
logger.Logger.WithField("legacy_app_count", legacyAppCount).Debug("Enqueuing publish of legacy info event")
|
||||
svc.nip47Service.EnqueueNip47InfoPublishRequest(0 /* unused */, svc.keys.GetNostrPublicKey(), svc.keys.GetNostrSecretKey())
|
||||
for _, relayUrl := range svc.cfg.GetRelayUrls() {
|
||||
svc.nip47Service.EnqueueNip47InfoPublishRequest(0 /* unused */, svc.keys.GetNostrPublicKey(), svc.keys.GetNostrSecretKey(), relayUrl)
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
|
|
@ -191,12 +154,14 @@ func (svc *service) publishAllAppInfoEvents() {
|
|||
return
|
||||
}
|
||||
logger.Logger.WithField("app_id", app.ID).Debug("Enqueuing publish of app info event")
|
||||
svc.nip47Service.EnqueueNip47InfoPublishRequest(app.ID, *app.WalletPubkey, walletPrivKey)
|
||||
for _, relayUrl := range svc.cfg.GetRelayUrls() {
|
||||
svc.nip47Service.EnqueueNip47InfoPublishRequest(app.ID, *app.WalletPubkey, walletPrivKey, relayUrl)
|
||||
}
|
||||
}(app)
|
||||
}
|
||||
}
|
||||
|
||||
func (svc *service) startAllExistingAppsWalletSubscriptions(ctx context.Context, relay *nostr.Relay) {
|
||||
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 {
|
||||
|
|
@ -206,7 +171,7 @@ func (svc *service) startAllExistingAppsWalletSubscriptions(ctx context.Context,
|
|||
|
||||
for _, app := range apps {
|
||||
go func(app db.App) {
|
||||
err := svc.startAppWalletSubscription(ctx, relay, *app.WalletPubkey)
|
||||
err := svc.startAppWalletSubscription(ctx, pool, *app.WalletPubkey)
|
||||
if err != nil && !errors.Is(err, context.Canceled) {
|
||||
logger.Logger.WithError(err).WithFields(logrus.Fields{
|
||||
"app_id": app.ID}).Error("Subscription error")
|
||||
|
|
@ -216,40 +181,48 @@ func (svc *service) startAllExistingAppsWalletSubscriptions(ctx context.Context,
|
|||
}
|
||||
}
|
||||
|
||||
func (svc *service) startAppWalletSubscription(ctx context.Context, relay *nostr.Relay, appWalletPubKey string) error {
|
||||
func (svc *service) startAppWalletSubscription(ctx context.Context, pool *nostr.SimplePool, appWalletPubKey string) error {
|
||||
|
||||
logger.Logger.Info("Subscribing to events for wallet ", appWalletPubKey)
|
||||
sub, err := relay.Subscribe(ctx, svc.createFilters(appWalletPubKey))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to subscribe to events: %w", err)
|
||||
|
||||
filter := nostr.Filter{
|
||||
Tags: nostr.TagMap{"p": []string{appWalletPubKey}},
|
||||
Kinds: []int{models.REQUEST_KIND},
|
||||
}
|
||||
|
||||
// register a subscriber for "nwc_app_deleted" events, which handles nostr subscription cancel and nip47 info event deletion
|
||||
deleteEventSubscriber := deleteAppConsumer{nostrSubscription: sub, walletPubkey: appWalletPubKey, svc: svc, relay: relay}
|
||||
svc.eventPublisher.RegisterSubscriber(&deleteEventSubscriber)
|
||||
subCtx, cancelSubscription := context.WithCancel(ctx)
|
||||
eventsChannel := pool.SubscribeMany(subCtx, svc.cfg.GetRelayUrls(), filter)
|
||||
|
||||
err = svc.StartSubscription(sub.Context, sub)
|
||||
svc.eventPublisher.RemoveSubscriber(&deleteEventSubscriber)
|
||||
// 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 {
|
||||
return fmt.Errorf("got an error from the relay while listening to subscription: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (svc *service) StartSubscription(ctx context.Context, sub *nostr.Subscription) error {
|
||||
func (svc *service) watchSubscription(ctx context.Context, pool *nostr.SimplePool, eventsChannel chan nostr.RelayEvent) error {
|
||||
go func() {
|
||||
// loop through incoming events
|
||||
for event := range sub.Events {
|
||||
go svc.nip47Service.HandleEvent(ctx, sub.Relay, event, svc.lnClient)
|
||||
for event := range eventsChannel {
|
||||
go svc.nip47Service.HandleEvent(ctx, pool, event.Event, svc.lnClient)
|
||||
}
|
||||
logger.Logger.Debug("Relay subscription events channel ended")
|
||||
}()
|
||||
|
||||
<-ctx.Done()
|
||||
|
||||
if err := sub.Relay.ConnectionError; err != nil {
|
||||
return fmt.Errorf("relay connection error: %w", err)
|
||||
}
|
||||
logger.Logger.Info("Exiting subscription...")
|
||||
return nil
|
||||
}
|
||||
|
|
@ -408,23 +381,6 @@ func (svc *service) launchLNBackend(ctx context.Context, encryptionKey string) e
|
|||
return nil
|
||||
}
|
||||
|
||||
func closeRelay(relay *nostr.Relay) {
|
||||
if relay != nil && relay.IsConnected() {
|
||||
logger.Logger.Info("Closing relay connection...")
|
||||
func() {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
logger.Logger.WithField("r", r).Error("Recovered from panic when closing relay")
|
||||
}
|
||||
}()
|
||||
err := relay.Close()
|
||||
if err != nil {
|
||||
logger.Logger.WithError(err).Error("Could not close relay connection")
|
||||
}
|
||||
}()
|
||||
}
|
||||
}
|
||||
|
||||
func (svc *service) requestVssToken(ctx context.Context) (string, error) {
|
||||
nodeLastStartTime, _ := svc.cfg.Get("NodeLastStartTime", "")
|
||||
|
||||
|
|
|
|||
|
|
@ -10,8 +10,7 @@ import (
|
|||
|
||||
type updateAppConsumer struct {
|
||||
events.EventSubscriber
|
||||
svc *service
|
||||
relay *nostr.Relay
|
||||
svc *service
|
||||
}
|
||||
|
||||
// When a app is updated, re-publish the nip47 info event
|
||||
|
|
@ -42,7 +41,10 @@ func (s *updateAppConsumer) ConsumeEvent(ctx context.Context, event *events.Even
|
|||
}
|
||||
|
||||
if s.svc.keys.GetNostrPublicKey() != walletPubKey {
|
||||
// only need to re-publish the nip47 event info if it is not a legacy wallet
|
||||
s.svc.nip47Service.EnqueueNip47InfoPublishRequest(id, walletPubKey, walletPrivKey)
|
||||
// only need to re-publish the nip47 event info if it is not a legacy app connection (shared wallet pubkey)
|
||||
// (legacy app connection can be used for multiple apps - so it cannot be app-specific)
|
||||
for _, relayUrl := range s.svc.cfg.GetRelayUrls() {
|
||||
s.svc.nip47Service.EnqueueNip47InfoPublishRequest(id, walletPubKey, walletPrivKey, relayUrl)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -7,16 +7,34 @@ import (
|
|||
"github.com/nbd-wtf/go-nostr"
|
||||
)
|
||||
|
||||
type mockRelay struct {
|
||||
type mockSimplePool struct {
|
||||
PublishedEvents []*nostr.Event
|
||||
}
|
||||
|
||||
func NewMockRelay() *mockRelay {
|
||||
return &mockRelay{}
|
||||
func NewMockSimplePool() *mockSimplePool {
|
||||
return &mockSimplePool{}
|
||||
}
|
||||
|
||||
func (relay *mockRelay) Publish(ctx context.Context, event nostr.Event) error {
|
||||
func (relay *mockSimplePool) PublishMany(ctx context.Context, relayUrls []string, event nostr.Event) chan nostr.PublishResult {
|
||||
logger.Logger.WithField("event", event).Info("Mock Publishing event")
|
||||
relay.PublishedEvents = append(relay.PublishedEvents, &event)
|
||||
|
||||
channel := make(chan nostr.PublishResult)
|
||||
go func() {
|
||||
channel <- nostr.PublishResult{
|
||||
RelayURL: "wss://fakerelay.com/v1",
|
||||
}
|
||||
close(channel)
|
||||
}()
|
||||
return channel
|
||||
}
|
||||
|
||||
func (relay *mockSimplePool) QuerySingle(
|
||||
ctx context.Context,
|
||||
urls []string,
|
||||
filter nostr.Filter,
|
||||
opts ...nostr.SubscriptionOption,
|
||||
) *nostr.RelayEvent {
|
||||
logger.Logger.Error("Mock pool QuerySingle is not supported yet")
|
||||
return nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -404,46 +404,48 @@ func (_c *MockConfig_GetNetwork_Call) RunAndReturn(run func() string) *MockConfi
|
|||
return _c
|
||||
}
|
||||
|
||||
// GetRelayUrl provides a mock function for the type MockConfig
|
||||
func (_mock *MockConfig) GetRelayUrl() string {
|
||||
// GetRelayUrls provides a mock function for the type MockConfig
|
||||
func (_mock *MockConfig) GetRelayUrls() []string {
|
||||
ret := _mock.Called()
|
||||
|
||||
if len(ret) == 0 {
|
||||
panic("no return value specified for GetRelayUrl")
|
||||
panic("no return value specified for GetRelayUrls")
|
||||
}
|
||||
|
||||
var r0 string
|
||||
if returnFunc, ok := ret.Get(0).(func() string); ok {
|
||||
var r0 []string
|
||||
if returnFunc, ok := ret.Get(0).(func() []string); ok {
|
||||
r0 = returnFunc()
|
||||
} else {
|
||||
r0 = ret.Get(0).(string)
|
||||
if ret.Get(0) != nil {
|
||||
r0 = ret.Get(0).([]string)
|
||||
}
|
||||
}
|
||||
return r0
|
||||
}
|
||||
|
||||
// MockConfig_GetRelayUrl_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'GetRelayUrl'
|
||||
type MockConfig_GetRelayUrl_Call struct {
|
||||
// MockConfig_GetRelayUrls_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'GetRelayUrls'
|
||||
type MockConfig_GetRelayUrls_Call struct {
|
||||
*mock.Call
|
||||
}
|
||||
|
||||
// GetRelayUrl is a helper method to define mock.On call
|
||||
func (_e *MockConfig_Expecter) GetRelayUrl() *MockConfig_GetRelayUrl_Call {
|
||||
return &MockConfig_GetRelayUrl_Call{Call: _e.mock.On("GetRelayUrl")}
|
||||
// GetRelayUrls is a helper method to define mock.On call
|
||||
func (_e *MockConfig_Expecter) GetRelayUrls() *MockConfig_GetRelayUrls_Call {
|
||||
return &MockConfig_GetRelayUrls_Call{Call: _e.mock.On("GetRelayUrls")}
|
||||
}
|
||||
|
||||
func (_c *MockConfig_GetRelayUrl_Call) Run(run func()) *MockConfig_GetRelayUrl_Call {
|
||||
func (_c *MockConfig_GetRelayUrls_Call) Run(run func()) *MockConfig_GetRelayUrls_Call {
|
||||
_c.Call.Run(func(args mock.Arguments) {
|
||||
run()
|
||||
})
|
||||
return _c
|
||||
}
|
||||
|
||||
func (_c *MockConfig_GetRelayUrl_Call) Return(s string) *MockConfig_GetRelayUrl_Call {
|
||||
_c.Call.Return(s)
|
||||
func (_c *MockConfig_GetRelayUrls_Call) Return(strings []string) *MockConfig_GetRelayUrls_Call {
|
||||
_c.Call.Return(strings)
|
||||
return _c
|
||||
}
|
||||
|
||||
func (_c *MockConfig_GetRelayUrl_Call) RunAndReturn(run func() string) *MockConfig_GetRelayUrl_Call {
|
||||
func (_c *MockConfig_GetRelayUrls_Call) RunAndReturn(run func() []string) *MockConfig_GetRelayUrls_Call {
|
||||
_c.Call.Return(run)
|
||||
return _c
|
||||
}
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ import (
|
|||
"github.com/getAlby/hub/config"
|
||||
"github.com/getAlby/hub/events"
|
||||
"github.com/getAlby/hub/lnclient"
|
||||
"github.com/getAlby/hub/service"
|
||||
"github.com/getAlby/hub/service/keys"
|
||||
"github.com/getAlby/hub/swaps"
|
||||
"github.com/getAlby/hub/transactions"
|
||||
|
|
@ -365,6 +366,52 @@ func (_c *MockService_GetLNClient_Call) RunAndReturn(run func() lnclient.LNClien
|
|||
return _c
|
||||
}
|
||||
|
||||
// GetRelayStatuses provides a mock function for the type MockService
|
||||
func (_mock *MockService) GetRelayStatuses() []service.RelayStatus {
|
||||
ret := _mock.Called()
|
||||
|
||||
if len(ret) == 0 {
|
||||
panic("no return value specified for GetRelayStatuses")
|
||||
}
|
||||
|
||||
var r0 []service.RelayStatus
|
||||
if returnFunc, ok := ret.Get(0).(func() []service.RelayStatus); ok {
|
||||
r0 = returnFunc()
|
||||
} else {
|
||||
if ret.Get(0) != nil {
|
||||
r0 = ret.Get(0).([]service.RelayStatus)
|
||||
}
|
||||
}
|
||||
return r0
|
||||
}
|
||||
|
||||
// MockService_GetRelayStatuses_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'GetRelayStatuses'
|
||||
type MockService_GetRelayStatuses_Call struct {
|
||||
*mock.Call
|
||||
}
|
||||
|
||||
// GetRelayStatuses is a helper method to define mock.On call
|
||||
func (_e *MockService_Expecter) GetRelayStatuses() *MockService_GetRelayStatuses_Call {
|
||||
return &MockService_GetRelayStatuses_Call{Call: _e.mock.On("GetRelayStatuses")}
|
||||
}
|
||||
|
||||
func (_c *MockService_GetRelayStatuses_Call) Run(run func()) *MockService_GetRelayStatuses_Call {
|
||||
_c.Call.Run(func(args mock.Arguments) {
|
||||
run()
|
||||
})
|
||||
return _c
|
||||
}
|
||||
|
||||
func (_c *MockService_GetRelayStatuses_Call) Return(relayStatuss []service.RelayStatus) *MockService_GetRelayStatuses_Call {
|
||||
_c.Call.Return(relayStatuss)
|
||||
return _c
|
||||
}
|
||||
|
||||
func (_c *MockService_GetRelayStatuses_Call) RunAndReturn(run func() []service.RelayStatus) *MockService_GetRelayStatuses_Call {
|
||||
_c.Call.Return(run)
|
||||
return _c
|
||||
}
|
||||
|
||||
// GetStartupState provides a mock function for the type MockService
|
||||
func (_mock *MockService) GetStartupState() string {
|
||||
ret := _mock.Called()
|
||||
|
|
@ -501,50 +548,6 @@ func (_c *MockService_GetTransactionsService_Call) RunAndReturn(run func() trans
|
|||
return _c
|
||||
}
|
||||
|
||||
// IsRelayReady provides a mock function for the type MockService
|
||||
func (_mock *MockService) IsRelayReady() bool {
|
||||
ret := _mock.Called()
|
||||
|
||||
if len(ret) == 0 {
|
||||
panic("no return value specified for IsRelayReady")
|
||||
}
|
||||
|
||||
var r0 bool
|
||||
if returnFunc, ok := ret.Get(0).(func() bool); ok {
|
||||
r0 = returnFunc()
|
||||
} else {
|
||||
r0 = ret.Get(0).(bool)
|
||||
}
|
||||
return r0
|
||||
}
|
||||
|
||||
// MockService_IsRelayReady_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'IsRelayReady'
|
||||
type MockService_IsRelayReady_Call struct {
|
||||
*mock.Call
|
||||
}
|
||||
|
||||
// IsRelayReady is a helper method to define mock.On call
|
||||
func (_e *MockService_Expecter) IsRelayReady() *MockService_IsRelayReady_Call {
|
||||
return &MockService_IsRelayReady_Call{Call: _e.mock.On("IsRelayReady")}
|
||||
}
|
||||
|
||||
func (_c *MockService_IsRelayReady_Call) Run(run func()) *MockService_IsRelayReady_Call {
|
||||
_c.Call.Run(func(args mock.Arguments) {
|
||||
run()
|
||||
})
|
||||
return _c
|
||||
}
|
||||
|
||||
func (_c *MockService_IsRelayReady_Call) Return(b bool) *MockService_IsRelayReady_Call {
|
||||
_c.Call.Return(b)
|
||||
return _c
|
||||
}
|
||||
|
||||
func (_c *MockService_IsRelayReady_Call) RunAndReturn(run func() bool) *MockService_IsRelayReady_Call {
|
||||
_c.Call.Return(run)
|
||||
return _c
|
||||
}
|
||||
|
||||
// Shutdown provides a mock function for the type MockService
|
||||
func (_mock *MockService) Shutdown() {
|
||||
_mock.Called()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue