Add relay repository

cmd/event-service/di/inject_adapters.go

@@ -37,6 +37,9 @@ var sqliteTestAdaptersSet = wire.NewSet(
 var sqliteTxAdaptersSet = wire.NewSet(
 	sqlite.NewEventRepository,
 	wire.Bind(new(app.EventRepository), new(*sqlite.EventRepository)),
+
+	sqlite.NewRelayRepository,
+	wire.Bind(new(app.RelayRepository), new(*sqlite.RelayRepository)),
 )
 
 var adaptersSet = wire.NewSet(

cmd/event-service/di/inject_pubsub.go

@@ -2,6 +2,7 @@ package di
 
 import (
 	"github.com/google/wire"
+	"github.com/planetary-social/nos-event-service/service/adapters/gcp"
 	"github.com/planetary-social/nos-event-service/service/adapters/memorypubsub"
 	"github.com/planetary-social/nos-event-service/service/adapters/sqlite"
 	"github.com/planetary-social/nos-event-service/service/app"
@@ -26,3 +27,8 @@ var sqliteTxPubsubSet = wire.NewSet(
 	sqlite.NewPublisher,
 	wire.Bind(new(app.Publisher), new(*sqlite.Publisher)),
 )
+
+var externalPubsubSet = wire.NewSet(
+	gcp.NewNoopPublisher,
+	wire.Bind(new(app.ExternalEventPublisher), new(*gcp.NoopPublisher)), // todo
+)

cmd/event-service/di/service.go

@@ -2,10 +2,10 @@ package di
 
 import (
 	"context"
-	"github.com/planetary-social/nos-event-service/internal/migrations"
 
 	"github.com/boreq/errors"
 	"github.com/hashicorp/go-multierror"
+	"github.com/planetary-social/nos-event-service/internal/migrations"
 	"github.com/planetary-social/nos-event-service/service/app"
 	"github.com/planetary-social/nos-event-service/service/ports/http"
 	"github.com/planetary-social/nos-event-service/service/ports/memorypubsub"

cmd/event-service/di/wire.go

@@ -32,6 +32,7 @@ func BuildService(context.Context, config.Config) (Service, func(), error) {
 		adaptersSet,
 		migrationsAdaptersSet,
 		domainSet,
+		externalPubsubSet,
 	)
 	return Service{}, nil, nil
 }

internal/fixtures/fixtures.go

@@ -154,6 +154,10 @@ func SomeBytesOfLen(l int) []byte {
 	return b
 }
 
+func SomeMaybeRelayAddress() domain.MaybeRelayAddress {
+	return domain.NewMaybeRelayAddress(SomeString())
+}
+
 var letters = []rune("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ")
 
 func randSeq(n int) string {

internal/migrations/migrations_test.go

@@ -3,12 +3,12 @@ package migrations_test
 import (
 	"context"
 	"fmt"
-	"github.com/planetary-social/nos-event-service/internal/migrations"
 	"testing"
 
 	"github.com/planetary-social/nos-event-service/internal"
 	"github.com/planetary-social/nos-event-service/internal/fixtures"
 	"github.com/planetary-social/nos-event-service/internal/logging"
+	"github.com/planetary-social/nos-event-service/internal/migrations"
 	"github.com/stretchr/testify/require"
 )
 

service/adapters/sqlite/migrations.go

@@ -3,9 +3,9 @@ package sqlite
 import (
 	"context"
 	"database/sql"
-	"github.com/planetary-social/nos-event-service/internal/migrations"
 
 	"github.com/boreq/errors"
+	"github.com/planetary-social/nos-event-service/internal/migrations"
 )
 
 func NewMigrations(fns *MigrationFns) (migrations.Migrations, error) {

service/adapters/sqlite/migrations_storage.go

@@ -3,9 +3,9 @@ package sqlite
 import (
 	"database/sql"
 	"encoding/json"
-	"github.com/planetary-social/nos-event-service/internal/migrations"
 
 	"github.com/boreq/errors"
+	"github.com/planetary-social/nos-event-service/internal/migrations"
 )
 
 type MigrationsStorage struct {

service/adapters/sqlite/migrations_storage_test.go

@@ -1,10 +1,10 @@
 package sqlite_test
 
 import (
-	"github.com/planetary-social/nos-event-service/internal/migrations"
 	"testing"
 
 	"github.com/planetary-social/nos-event-service/internal/fixtures"
+	"github.com/planetary-social/nos-event-service/internal/migrations"
 	"github.com/stretchr/testify/require"
 )
 

service/adapters/sqlite/relay_repository.go

@@ -0,0 +1,78 @@
+package sqlite
+
+import (
+	"context"
+	"database/sql"
+
+	"github.com/boreq/errors"
+	"github.com/planetary-social/nos-event-service/service/domain"
+)
+
+type RelayRepository struct {
+	tx *sql.Tx
+}
+
+func NewRelayRepository(tx *sql.Tx) *RelayRepository {
+	return &RelayRepository{tx: tx}
+}
+
+func (r *RelayRepository) Save(ctx context.Context, eventID domain.EventId, relayAddress domain.MaybeRelayAddress) error {
+	_, err := r.tx.Exec(`
+	INSERT OR IGNORE INTO relays(address)
+	VALUES($1)`,
+		relayAddress.String(),
+	)
+	if err != nil {
+		return errors.Wrap(err, "error inserting the relay address")
+	}
+
+	row := r.tx.QueryRow(`
+	SELECT id FROM relays
+	WHERE address=$1`,
+		relayAddress.String(),
+	)
+	var dbRelayId int
+	if err := row.Scan(&dbRelayId); err != nil {
+		return errors.Wrap(err, "error getting the relay id")
+
+	}
+
+	row = r.tx.QueryRow(`
+	SELECT id FROM events
+	WHERE event_id=$1`,
+		eventID.Hex(),
+	)
+	var dbEventId int
+	if err := row.Scan(&dbEventId); err != nil {
+		return errors.Wrap(err, "error getting the event id")
+	}
+
+	_, err = r.tx.Exec(`
+	INSERT OR IGNORE INTO events_to_relays(event_id, relay_id)
+	VALUES($1, $2)`,
+		dbEventId, dbRelayId,
+	)
+	if err != nil {
+		return errors.Wrap(err, "error inserting the relationship")
+	}
+
+	return nil
+}
+
+func (r *RelayRepository) List(ctx context.Context) ([]domain.MaybeRelayAddress, error) {
+	rows, err := r.tx.Query(`SELECT address FROM relays`)
+	if err != nil {
+		return nil, errors.Wrap(err, "error selecting addresses")
+	}
+
+	var result []domain.MaybeRelayAddress
+	for rows.Next() {
+		var addressString string
+		if err := rows.Scan(&addressString); err != nil {
+			return nil, errors.Wrap(err, "error scanning")
+		}
+		result = append(result, domain.NewMaybeRelayAddress(addressString))
+	}
+
+	return result, nil
+}