Implement proper backoff to handle arbitrarily failing messages

cmd/crossposting-service/di/inject_adapters.go

@@ -22,8 +22,6 @@ var sqliteAdaptersSet = wire.NewSet(
 
 	newAdaptersFactoryFn,
 
-	sqlite.NewMigrations,
-
 	wire.Struct(new(buildTransactionSqliteAdaptersDependencies), "*"),
 )
 
@@ -34,8 +32,6 @@ var sqliteTestAdaptersSet = wire.NewSet(
 
 	newTestAdaptersFactoryFn,
 
-	sqlite.NewMigrations,
-
 	wire.Struct(new(buildTransactionSqliteAdaptersDependencies), "*"),
 )
 

cmd/crossposting-service/di/inject_migrations.go

@@ -0,0 +1,20 @@
+package di
+
+import (
+	"github.com/google/wire"
+	"github.com/planetary-social/nos-crossposting-service/migrations"
+	"github.com/planetary-social/nos-crossposting-service/service/adapters"
+	"github.com/planetary-social/nos-crossposting-service/service/adapters/sqlite"
+)
+
+var migrationsAdaptersSet = wire.NewSet(
+	sqlite.NewMigrations,
+	sqlite.NewMigrationFns,
+	migrations.NewRunner,
+
+	sqlite.NewMigrationsStorage,
+	wire.Bind(new(migrations.Storage), new(*sqlite.MigrationsStorage)),
+
+	adapters.NewLoggingMigrationsProgressCallback,
+	wire.Bind(new(migrations.ProgressCallback), new(*adapters.LoggingMigrationsProgressCallback)),
+)

cmd/crossposting-service/di/inject_pubsub.go

@@ -1,7 +1,6 @@
 package di
 
 import (
-	watermillsql "github.com/ThreeDotsLabs/watermill-sql/v2/pkg/sql"
 	"github.com/google/wire"
 	"github.com/planetary-social/nos-crossposting-service/service/adapters/memorypubsub"
 	"github.com/planetary-social/nos-crossposting-service/service/adapters/sqlite"
@@ -16,17 +15,12 @@ var memoryPubsubSet = wire.NewSet(
 )
 
 var sqlitePubsubSet = wire.NewSet(
-	sqlite.NewSqliteSchema,
-	wire.Bind(new(watermillsql.SchemaAdapter), new(sqlite.SqliteSchema)),
-
-	sqlite.NewWatermillOffsetsAdapter,
-	sqlite.NewWatermillSubscriber,
 	sqlitepubsubport.NewTweetCreatedEventSubscriber,
 	sqlite.NewSubscriber,
+	sqlite.NewPubSub,
 )
 
 var sqliteTxPubsubSet = wire.NewSet(
-	sqlite.NewWatermillPublisher,
 	sqlite.NewPublisher,
 	wire.Bind(new(app.Publisher), new(*sqlite.Publisher)),
 )

cmd/crossposting-service/di/service.go

@@ -5,7 +5,7 @@ import (
 
 	"github.com/boreq/errors"
 	"github.com/hashicorp/go-multierror"
-	"github.com/planetary-social/nos-crossposting-service/service/adapters/sqlite"
+	"github.com/planetary-social/nos-crossposting-service/migrations"
 	"github.com/planetary-social/nos-crossposting-service/service/app"
 	"github.com/planetary-social/nos-crossposting-service/service/ports/http"
 	"github.com/planetary-social/nos-crossposting-service/service/ports/memorypubsub"
@@ -19,7 +19,9 @@ type Service struct {
 	downloader                  *app.Downloader
 	receivedEventSubscriber     *memorypubsub.ReceivedEventSubscriber
 	tweetCreatedEventSubscriber *sqlitepubsub.TweetCreatedEventSubscriber
-	migrations                  *sqlite.Migrations
+	migrationsRunner            *migrations.Runner
+	migrations                  migrations.Migrations
+	migrationsProgressCallback  migrations.ProgressCallback
 }
 
 func NewService(
@@ -29,7 +31,9 @@ func NewService(
 	downloader *app.Downloader,
 	receivedEventSubscriber *memorypubsub.ReceivedEventSubscriber,
 	tweetCreatedEventSubscriber *sqlitepubsub.TweetCreatedEventSubscriber,
-	migrations *sqlite.Migrations,
+	migrationsRunner *migrations.Runner,
+	migrations migrations.Migrations,
+	migrationsProgressCallback migrations.ProgressCallback,
 ) Service {
 	return Service{
 		app:                         app,
@@ -38,7 +42,9 @@ func NewService(
 		downloader:                  downloader,
 		receivedEventSubscriber:     receivedEventSubscriber,
 		tweetCreatedEventSubscriber: tweetCreatedEventSubscriber,
+		migrationsRunner:            migrationsRunner,
 		migrations:                  migrations,
+		migrationsProgressCallback:  migrationsProgressCallback,
 	}
 }
 
@@ -47,7 +53,7 @@ func (s Service) App() app.Application {
 }
 
 func (s Service) ExecuteMigrations(ctx context.Context) error {
-	return s.migrations.Execute(ctx)
+	return s.migrationsRunner.Run(ctx, s.migrations, s.migrationsProgressCallback)
 }
 
 func (s Service) Run(ctx context.Context) error {

cmd/crossposting-service/di/wire.go

@@ -8,7 +8,6 @@ import (
 	"database/sql"
 	"testing"
 
-	"github.com/ThreeDotsLabs/watermill"
 	"github.com/google/wire"
 	"github.com/planetary-social/nos-crossposting-service/internal/fixtures"
 	"github.com/planetary-social/nos-crossposting-service/internal/logging"
@@ -31,6 +30,7 @@ func BuildService(context.Context, config.Config) (Service, func(), error) {
 		loggingSet,
 		adaptersSet,
 		tweetGeneratorSet,
+		migrationsAdaptersSet,
 	)
 	return Service{}, nil, nil
 }
@@ -43,6 +43,7 @@ func BuildTestAdapters(context.Context, testing.TB) (sqlite.TestedItems, func(),
 		sqlitePubsubSet,
 		loggingSet,
 		newTestAdaptersConfig,
+		migrationsAdaptersSet,
 	)
 	return sqlite.TestedItems{}, nil, nil
 }
@@ -61,13 +62,13 @@ func newTestAdaptersConfig(tb testing.TB) (config.Config, error) {
 }
 
 type buildTransactionSqliteAdaptersDependencies struct {
-	LoggerAdapter watermill.LoggerAdapter
+	Logger logging.Logger
 }
 
 func buildTransactionSqliteAdapters(*sql.DB, *sql.Tx, buildTransactionSqliteAdaptersDependencies) (app.Adapters, error) {
 	wire.Build(
 		wire.Struct(new(app.Adapters), "*"),
-		wire.FieldsOf(new(buildTransactionSqliteAdaptersDependencies), "LoggerAdapter"),
+		wire.FieldsOf(new(buildTransactionSqliteAdaptersDependencies), "Logger"),
 
 		sqliteTxAdaptersSet,
 		sqliteTxPubsubSet,
@@ -79,7 +80,7 @@ func buildTransactionSqliteAdapters(*sql.DB, *sql.Tx, buildTransactionSqliteAdap
 func buildTestTransactionSqliteAdapters(*sql.DB, *sql.Tx, buildTransactionSqliteAdaptersDependencies) (sqlite.TestAdapters, error) {
 	wire.Build(
 		wire.Struct(new(sqlite.TestAdapters), "*"),
-		wire.FieldsOf(new(buildTransactionSqliteAdaptersDependencies), "LoggerAdapter"),
+		wire.FieldsOf(new(buildTransactionSqliteAdaptersDependencies), "Logger"),
 
 		sqliteTxAdaptersSet,
 		sqliteTxPubsubSet,

go.mod

@@ -4,7 +4,6 @@ go 1.21
 
 require (
 	github.com/ThreeDotsLabs/watermill v1.3.1
-	github.com/ThreeDotsLabs/watermill-sql/v2 v2.0.0
 	github.com/boreq/errors v0.1.0
 	github.com/boreq/rest v0.1.0
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1
@@ -19,7 +18,6 @@ require (
 	github.com/mattn/go-sqlite3 v1.14.17
 	github.com/nbd-wtf/go-nostr v0.18.10
 	github.com/oklog/ulid/v2 v2.1.0
-	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.16.0
 	github.com/rs/zerolog v1.29.1
 	github.com/sirupsen/logrus v1.9.3
@@ -58,6 +56,7 @@ require (
 	github.com/prometheus/common v0.42.0 // indirect
 	github.com/prometheus/procfs v0.10.1 // indirect
 	github.com/puzpuzpuz/xsync v1.5.2 // indirect
+	github.com/rogpeppe/go-internal v1.10.0 // indirect
 	github.com/tidwall/gjson v1.14.4 // indirect
 	github.com/tidwall/match v1.1.1 // indirect
 	github.com/tidwall/pretty v1.2.0 // indirect