Time window based replication

Makefile

@@ -17,6 +17,10 @@ fmt:
 test:
 	go test -race ./...
 
+.PHONY: test-nocache
+test-nocache:
+	go test -race -count=1 ./...
+
 .PHONY: test-bench
 test-bench:
 	go test -v -race -run="^$$" -bench=. -benchtime=1x ./...

cmd/event-service/di/inject_adapters.go

@@ -6,6 +6,7 @@ import (
 	"github.com/boreq/errors"
 	"github.com/google/wire"
 	"github.com/planetary-social/nos-event-service/internal/logging"
+	"github.com/planetary-social/nos-event-service/service/adapters"
 	"github.com/planetary-social/nos-event-service/service/adapters/prometheus"
 	"github.com/planetary-social/nos-event-service/service/adapters/sqlite"
 	"github.com/planetary-social/nos-event-service/service/app"
@@ -64,6 +65,9 @@ var adaptersSet = wire.NewSet(
 	wire.Bind(new(app.Metrics), new(*prometheus.Prometheus)),
 	wire.Bind(new(relays.Metrics), new(*prometheus.Prometheus)),
 	wire.Bind(new(downloader.Metrics), new(*prometheus.Prometheus)),
+
+	adapters.NewCurrentTimeProvider,
+	wire.Bind(new(downloader.CurrentTimeProvider), new(*adapters.CurrentTimeProvider)),
 )
 
 func newAdaptersFactoryFn(deps buildTransactionSqliteAdaptersDependencies) sqlite.AdaptersFactoryFn {

cmd/event-service/di/service.go

@@ -23,6 +23,7 @@ type Service struct {
 	eventSavedEventSubscriber  *sqlitepubsub.EventSavedEventSubscriber
 	metricsTimer               *timer.Metrics
 	transactionRunner          *sqlite.TransactionRunner
+	taskScheduler              *downloader.TaskScheduler
 	migrationsRunner           *migrations.Runner
 	migrations                 migrations.Migrations
 	migrationsProgressCallback migrations.ProgressCallback
@@ -36,6 +37,7 @@ func NewService(
 	eventSavedEventSubscriber *sqlitepubsub.EventSavedEventSubscriber,
 	metricsTimer *timer.Metrics,
 	transactionRunner *sqlite.TransactionRunner,
+	taskScheduler *downloader.TaskScheduler,
 	migrationsRunner *migrations.Runner,
 	migrations migrations.Migrations,
 	migrationsProgressCallback migrations.ProgressCallback,
@@ -49,6 +51,7 @@ func NewService(
 		metricsTimer:               metricsTimer,
 		transactionRunner:          transactionRunner,
 		migrationsRunner:           migrationsRunner,
+		taskScheduler:              taskScheduler,
 		migrations:                 migrations,
 		migrationsProgressCallback: migrationsProgressCallback,
 	}
@@ -99,6 +102,11 @@ func (s Service) Run(ctx context.Context) error {
 		errCh <- errors.Wrap(s.transactionRunner.Run(ctx), "transaction runner error")
 	}()
 
+	runners++
+	go func() {
+		errCh <- errors.Wrap(s.taskScheduler.Run(ctx), "task scheduler error")
+	}()
+
 	var err error
 	for i := 0; i < runners; i++ {
 		err = multierror.Append(err, errors.Wrap(<-errCh, "error returned by runner"))

cmd/event-service/di/wire.go

@@ -150,12 +150,20 @@ var downloaderSet = wire.NewSet(
 	wire.Bind(new(downloader.RelayConnections), new(*relays.RelayConnections)),
 
 	app.NewDatabasePublicKeySource,
-	wire.Bind(new(downloader.PublicKeySource), new(*app.DatabasePublicKeySource)),
+	newCachedPublicKeySource,
+	wire.Bind(new(downloader.PublicKeySource), new(*app.CachedDatabasePublicKeySource)),
 
 	relays.NewEventSender,
 	wire.Bind(new(app.EventSender), new(*relays.EventSender)),
+
+	downloader.NewTaskScheduler,
+	wire.Bind(new(downloader.Scheduler), new(*downloader.TaskScheduler)),
 )
 
+func newCachedPublicKeySource(underlying *app.DatabasePublicKeySource) *app.CachedDatabasePublicKeySource {
+	return app.NewCachedDatabasePublicKeySource(underlying)
+}
+
 var domainSet = wire.NewSet(
 	domain.NewRelaysExtractor,
 	wire.Bind(new(app.RelaysExtractor), new(*domain.RelaysExtractor)),

internal/fixtures/fixtures.go

@@ -8,11 +8,13 @@ import (
 	"math/rand"
 	"os"
 	"testing"
+	"time"
 
 	"github.com/nbd-wtf/go-nostr"
 	"github.com/planetary-social/nos-event-service/internal"
 	"github.com/planetary-social/nos-event-service/internal/logging"
 	"github.com/planetary-social/nos-event-service/service/domain"
+	"github.com/planetary-social/nos-event-service/service/domain/downloader"
 	"github.com/stretchr/testify/require"
 )
 
@@ -186,6 +188,18 @@ func SomeMaybeRelayAddress() domain.MaybeRelayAddress {
 	return domain.NewMaybeRelayAddress(SomeString())
 }
 
+func SomeTimeWindow() downloader.TimeWindow {
+	return downloader.MustNewTimeWindow(SomeTime(), SomeDuration())
+}
+
+func SomeTime() time.Time {
+	return time.Unix(int64(rand.Intn(10000000)), 0)
+}
+
+func SomeDuration() time.Duration {
+	return time.Duration(1+rand.Intn(100)) * time.Second
+}
+
 var letters = []rune("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ")
 
 func randSeq(n int) string {

service/adapters/current_time_provider.go

@@ -0,0 +1,16 @@
+package adapters
+
+import (
+	"time"
+)
+
+type CurrentTimeProvider struct {
+}
+
+func NewCurrentTimeProvider() *CurrentTimeProvider {
+	return &CurrentTimeProvider{}
+}
+
+func (c *CurrentTimeProvider) GetCurrentTime() time.Time {
+	return time.Now()
+}

service/adapters/mocks/current_time_provider.go

@@ -0,0 +1,21 @@
+package mocks
+
+import (
+	"time"
+)
+
+type CurrentTimeProvider struct {
+	t time.Time
+}
+
+func NewCurrentTimeProvider() *CurrentTimeProvider {
+	return &CurrentTimeProvider{}
+}
+
+func (c *CurrentTimeProvider) SetCurrentTime(t time.Time) {
+	c.t = t
+}
+
+func (c *CurrentTimeProvider) GetCurrentTime() time.Time {
+	return c.t
+}

service/app/sources.go

@@ -11,6 +11,8 @@ import (
 	"github.com/planetary-social/nos-event-service/service/domain/downloader"
 )
 
+const cachePublicKeysFor = 1 * time.Minute
+
 type DatabaseRelaySource struct {
 	transactionProvider TransactionProvider
 	logger              logging.Logger
@@ -110,3 +112,30 @@ func (d *DatabasePublicKeySource) GetPublicKeys(ctx context.Context) (downloader
 		publicKeysToMonitorFollowees.List(),
 	), nil
 }
+
+type CachedDatabasePublicKeySource struct {
+	keys   *downloader.PublicKeys
+	t      time.Time
+	source downloader.PublicKeySource
+}
+
+func NewCachedDatabasePublicKeySource(
+	source downloader.PublicKeySource,
+) *CachedDatabasePublicKeySource {
+	return &CachedDatabasePublicKeySource{
+		source: source,
+	}
+}
+
+func (d *CachedDatabasePublicKeySource) GetPublicKeys(ctx context.Context) (downloader.PublicKeys, error) {
+	if d.keys == nil || time.Since(d.t) > cachePublicKeysFor {
+		newKeys, err := d.source.GetPublicKeys(ctx)
+		if err != nil {
+			return downloader.PublicKeys{}, errors.Wrap(err, "error getting new public keys")
+		}
+		d.keys = &newKeys
+		d.t = time.Now()
+	}
+
+	return *d.keys, nil
+}