Discovery: update client copy of discovery service

discovery/client.go

@@ -0,0 +1,99 @@
+/*
+ * Copyright (C) 2024 Nuts community
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ *
+ */
+
+package discovery
+
+import (
+	"context"
+	"fmt"
+	"github.com/nuts-foundation/nuts-node/discovery/api/v1/client"
+	"github.com/nuts-foundation/nuts-node/discovery/log"
+	"time"
+)
+
+// clientUpdater is responsible for updating the presentations for the given services, at the given interval.
+// Callers should only call update().
+type clientUpdater struct {
+	services map[string]ServiceDefinition
+	store    *sqlStore
+	client   client.HTTPClient
+	verifier registrationVerifier
+}
+
+func newClientUpdater(services map[string]ServiceDefinition, store *sqlStore, verifier registrationVerifier, client client.HTTPClient) *clientUpdater {
+	return &clientUpdater{
+		services: services,
+		store:    store,
+		client:   client,
+		verifier: verifier,
+	}
+}
+
+// update starts a blocking loop that updates the presentations for the given services, at the given interval.
+// It returns when the context is cancelled.
+func (u *clientUpdater) update(ctx context.Context, interval time.Duration) {
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	u.doUpdate(ctx)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			u.doUpdate(ctx)
+		}
+	}
+}
+
+func (u *clientUpdater) doUpdate(ctx context.Context) {
+	for _, service := range u.services {
+		if err := u.updateService(ctx, service); err != nil {
+			log.Logger().Errorf("Failed to update service (id=%s): %s", service.ID, err)
+		}
+	}
+}
+
+func (u *clientUpdater) updateService(ctx context.Context, service ServiceDefinition) error {
+	currentTag, err := u.store.getTag(service.ID)
+	if err != nil {
+		return err
+	}
+	// convert *Tag to *string
+	var currentTagStr *string
+	if !currentTag.Empty() {
+		currentTagStr = new(string)
+		*currentTagStr = string(currentTag)
+	}
+	presentations, newTag, err := u.client.Get(ctx, service.Endpoint, currentTagStr)
+	if err != nil {
+		return fmt.Errorf("failed to get presentations from discovery service (id=%s): %w", service.ID, err)
+	}
+	// *string back to *Tag
+	newTagStr := new(Tag)
+	*newTagStr = Tag(*newTag)
+	for _, presentation := range presentations {
+		if err := u.verifier.verifyRegistration(service, presentation); err != nil {
+			log.Logger().WithError(err).Warnf("Presentation verification failed, not adding it (service=%s, id=%s)", service.ID, presentation.ID)
+			continue
+		}
+		if err := u.store.add(service.ID, presentation, newTagStr); err != nil {
+			return fmt.Errorf("failed to store presentation (service=%s, id=%s): %w", service.ID, presentation.ID, err)
+		}
+	}
+	return nil
+}

discovery/client_test.go

@@ -0,0 +1,138 @@
+/*
+ * Copyright (C) 2024 Nuts community
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ *
+ */
+
+package discovery
+
+import (
+	"context"
+	"errors"
+	"github.com/nuts-foundation/go-did/vc"
+	"github.com/nuts-foundation/nuts-node/discovery/api/v1/client"
+	"github.com/nuts-foundation/nuts-node/storage"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
+	"sync"
+	"testing"
+	"time"
+)
+
+func Test_clientUpdater_updateService(t *testing.T) {
+	storageEngine := storage.NewTestStorageEngine(t)
+	require.NoError(t, storageEngine.Start())
+	store, err := newSQLStore(storageEngine.GetSQLDatabase(), testDefinitions(), nil)
+	require.NoError(t, err)
+	ctx := context.Background()
+	newTag := "test"
+	serviceDefinition := testDefinitions()[testServiceID]
+
+	t.Run("no updates", func(t *testing.T) {
+		resetStore(t, storageEngine.GetSQLDatabase())
+		ctrl := gomock.NewController(t)
+		verifier := NewMockregistrationVerifier(ctrl)
+		httpClient := client.NewMockHTTPClient(ctrl)
+		updater := newClientUpdater(testDefinitions(), store, verifier, httpClient)
+
+		httpClient.EXPECT().Get(ctx, testDefinitions()[testServiceID].Endpoint, nil).Return([]vc.VerifiablePresentation{}, &newTag, nil)
+
+		err := updater.updateService(ctx, testDefinitions()[testServiceID])
+
+		require.NoError(t, err)
+	})
+	t.Run("updates", func(t *testing.T) {
+		resetStore(t, storageEngine.GetSQLDatabase())
+		ctrl := gomock.NewController(t)
+		verifier := NewMockregistrationVerifier(ctrl)
+		httpClient := client.NewMockHTTPClient(ctrl)
+		updater := newClientUpdater(testDefinitions(), store, verifier, httpClient)
+
+		httpClient.EXPECT().Get(ctx, serviceDefinition.Endpoint, nil).Return([]vc.VerifiablePresentation{vpAlice}, &newTag, nil)
+		verifier.EXPECT().verifyRegistration(serviceDefinition, vpAlice).Return(nil)
+
+		err := updater.updateService(ctx, testDefinitions()[testServiceID])
+
+		require.NoError(t, err)
+	})
+	t.Run("ignores invalid presentations", func(t *testing.T) {
+		resetStore(t, storageEngine.GetSQLDatabase())
+		ctrl := gomock.NewController(t)
+		verifier := NewMockregistrationVerifier(ctrl)
+		httpClient := client.NewMockHTTPClient(ctrl)
+		updater := newClientUpdater(testDefinitions(), store, verifier, httpClient)
+
+		httpClient.EXPECT().Get(ctx, serviceDefinition.Endpoint, nil).Return([]vc.VerifiablePresentation{vpAlice, vpBob}, &newTag, nil)
+		verifier.EXPECT().verifyRegistration(serviceDefinition, vpAlice).Return(errors.New("invalid presentation"))
+		verifier.EXPECT().verifyRegistration(serviceDefinition, vpBob).Return(nil)
+
+		err := updater.updateService(ctx, testDefinitions()[testServiceID])
+
+		require.NoError(t, err)
+		// Bob's VP should exist, Alice's not
+		exists, err := store.exists(testServiceID, bobDID.String(), vpBob.ID.String())
+		require.NoError(t, err)
+		require.True(t, exists)
+		exists, err = store.exists(testServiceID, aliceDID.String(), vpAlice.ID.String())
+		require.NoError(t, err)
+		require.False(t, exists)
+	})
+	t.Run("pass tag", func(t *testing.T) {
+		resetStore(t, storageEngine.GetSQLDatabase())
+		ctrl := gomock.NewController(t)
+		verifier := NewMockregistrationVerifier(ctrl)
+		httpClient := client.NewMockHTTPClient(ctrl)
+		inputTag := Tag("test")
+		_, err := store.updateTag(store.db, testServiceID, &inputTag)
+		require.NoError(t, err)
+		updater := newClientUpdater(testDefinitions(), store, verifier, httpClient)
+
+		clientTagStr := string(inputTag)
+		httpClient.EXPECT().Get(ctx, serviceDefinition.Endpoint, gomock.Eq(&clientTagStr)).Return([]vc.VerifiablePresentation{vpAlice}, &newTag, nil)
+		verifier.EXPECT().verifyRegistration(serviceDefinition, vpAlice).Return(nil)
+
+		err = updater.updateService(ctx, testDefinitions()[testServiceID])
+
+		require.NoError(t, err)
+	})
+}
+
+func Test_clientUpdater_update(t *testing.T) {
+	storageEngine := storage.NewTestStorageEngine(t)
+	require.NoError(t, storageEngine.Start())
+
+	t.Run("context cancel stops the loop", func(t *testing.T) {
+		store := setupStore(t, storageEngine.GetSQLDatabase())
+		ctrl := gomock.NewController(t)
+		verifier := NewMockregistrationVerifier(ctrl)
+		httpClient := client.NewMockHTTPClient(ctrl)
+		newTag := "test"
+		httpClient.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Return([]vc.VerifiablePresentation{}, &newTag, nil).MinTimes(1)
+		updater := newClientUpdater(testDefinitions(), store, verifier, httpClient)
+
+		ctx, cancel := context.WithCancel(context.Background())
+		wg := &sync.WaitGroup{}
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			updater.update(ctx, time.Millisecond)
+		}()
+		// make sure the loop has at least once
+		time.Sleep(5 * time.Millisecond)
+		// Make sure the function exits when the context is cancelled
+		cancel()
+		wg.Wait()
+	})
+}

discovery/cmd/cmd.go

@@ -33,5 +33,7 @@ func FlagSet() *pflag.FlagSet {
 	flagSet.StringSlice("discovery.server.definition_ids", defs.Server.DefinitionIDs,
 		"IDs of the Discovery Service Definitions for which to act as server. "+
 			"If an ID does not map to a loaded service definition, the node will fail to start.")
+	flagSet.Duration("discovery.client.update_interval", defs.Client.UpdateInterval, "How often to check for Discovery Services updates, "+
+		"specified as Golang duration (e.g. 1m, 1h30m).")
 	return flagSet
 }

discovery/config.go

@@ -18,9 +18,12 @@
 
 package discovery
 
+import "time"
+
 // Config holds the config of the module
 type Config struct {
 	Server      ServerConfig             `koanf:"server"`
+	Client      ClientConfig             `koanf:"client"`
 	Definitions ServiceDefinitionsConfig `koanf:"definitions"`
 }
 
@@ -35,10 +38,19 @@ type ServerConfig struct {
 	DefinitionIDs []string `koanf:"definition_ids"`
 }
 
+// ClientConfig holds the config for the client
+type ClientConfig struct {
+	// UpdateInterval specifies how often the client should update the Discovery Services.
+	UpdateInterval time.Duration `koanf:"update_interval"`
+}
+
 // DefaultConfig returns the default configuration.
 func DefaultConfig() Config {
 	return Config{
 		Server: ServerConfig{},
+		Client: ClientConfig{
+			UpdateInterval: 1 * time.Minute,
+		},
 	}
 }
 

discovery/interface.go

@@ -108,3 +108,7 @@ type SearchResult struct {
 	// It only includes constraint fields that have an ID.
 	Fields map[string]interface{} `json:"fields"`
 }
+
+type registrationVerifier interface {
+	verifyRegistration(definition ServiceDefinition, presentation vc.VerifiablePresentation) error
+}