fix: incremental migrations

internal/storage/bucket/migrations/12-moves-fill-transaction-id/up.sql

@@ -1,9 +1,9 @@
-set search_path = '{{.Bucket}}';
-
 do $$
 	declare
 		_batch_size integer := 30;
 	begin
+		set search_path = '{{ .Bucket }}';
+
 		loop
 			with _outdated_moves as (
 				select *
@@ -21,9 +21,13 @@ do $$
 			where moves.seq in (_outdated_moves.seq);
 
 			exit when not found;
+
+			raise info 'commit batch';
+			commit ;
 		end loop;
-	end
-$$;
 
-alter table moves
-alter column transactions_id set not null;
+		alter table moves
+		alter column transactions_id set not null;
+	end
+$$
+language plpgsql;

internal/storage/bucket/migrations/13-transactions-fill-inserted-at/up.sql

@@ -1,20 +1,20 @@
-set search_path = '{{.Bucket}}';
-
 do $$
 	declare
 		_batch_size integer := 30;
-		-- select the date where the "11-make-stateless" migration has been applied
-		_date timestamp without time zone = (
-			select tstamp
-			from _system.goose_db_version
-			where version_id = 12
-		);
-		_count integer = (
-			select count(*)
-			from logs
-			where date <= _date
-		);
+		_date timestamp without time zone;
+		_count integer;
 	begin
+		set search_path = '{{ .Bucket }}';
+
+		-- select the date where the "11-make-stateless" migration has been applied
+		select tstamp into _date
+		from _system.goose_db_version
+		where version_id = 12;
+
+		select count(*) into _count
+		from logs
+		where date <= _date;
+
 		for i in 0.._count by _batch_size loop
 			update transactions
 			set inserted_at = (
@@ -23,9 +23,8 @@ do $$
 				where transactions.id = (data->'transaction'->>'id')::bigint and transactions.ledger = ledger
 			)
 			where id >= i and id < i + _batch_size;
+
+			commit;
 		end loop;
 	end
-$$;
-
-alter table moves
-alter column transactions_id set not null;
+$$;

internal/storage/bucket/migrations/14-transactions-fill-pcv/up.sql

@@ -1,9 +1,9 @@
-set search_path = '{{.Bucket}}';
-
 do $$
 	declare
 		_batch_size integer := 30;
 	begin
+		set search_path = '{{ .Bucket }}';
+
 		loop
 			with _outdated_transactions as (
 				select id
@@ -37,8 +37,8 @@ do $$
 
 			exit when not found;
 		end loop;
+
+		alter table transactions
+		alter column post_commit_volumes set not null;
 	end
-$$;
-
-alter table transactions
-alter column post_commit_volumes set not null ;
+$$;

internal/storage/bucket/migrations/15-accounts-volumes-fill-history/up.sql

@@ -1,9 +1,9 @@
-set search_path = '{{.Bucket}}';
-
 do $$
 	declare
 		_missing record;
 	begin
+		set search_path = '{{ .Bucket }}';
+
 		loop
 			select distinct on (ledger, accounts_address, asset)
 				ledger,
@@ -35,6 +35,8 @@ do $$
 			(_missing.post_commit_volumes).outputs
 			)
 			on conflict do nothing; -- can be inserted by a concurrent transaction
+
+			commit;
 		end loop;
 	end
 $$;

internal/storage/bucket/migrations/16-transactions-metadata-fill-transaction-id/up.sql

@@ -1,9 +1,10 @@
-set search_path = '{{.Bucket}}';
 
 do $$
 	declare
 		_batch_size integer := 30;
 	begin
+		set search_path = '{{.Bucket}}';
+
 		loop
 			with _outdated_transactions_metadata as (
 				select seq
@@ -21,9 +22,12 @@ do $$
 			where transactions_metadata.seq in (_outdated_transactions_metadata.seq);
 
 			exit when not found;
+
+			commit ;
 		end loop;
+
+		alter table transactions_metadata
+		alter column transactions_id set not null ;
 	end
 $$;
 
-alter table transactions_metadata
-alter column transactions_id set not null ;

internal/storage/bucket/migrations/17-accounts-metadata-fill-address/up.sql

@@ -1,9 +1,10 @@
-set search_path = '{{.Bucket}}';
 
 do $$
 	declare
 		_batch_size integer := 30;
 	begin
+		set search_path = '{{.Bucket}}';
+
 		loop
 			with _outdated_accounts_metadata as (
 				select seq
@@ -21,9 +22,12 @@ do $$
 			where accounts_metadata.seq in (_outdated_accounts_metadata.seq);
 
 			exit when not found;
+
+			commit ;
 		end loop;
+
+		alter table accounts_metadata
+		alter column accounts_address set not null ;
 	end
 $$;
 
-alter table accounts_metadata
-alter column accounts_address set not null ;

internal/storage/bucket/migrations/18-logs-fill-memento/up.sql

@@ -1,9 +1,9 @@
-set search_path = '{{.Bucket}}';
-
 do $$
 	declare
 		_batch_size integer := 30;
 	begin
+		set search_path = '{{.Bucket}}';
+
 		loop
 			with _outdated_logs as (
 				select seq
@@ -17,9 +17,12 @@ do $$
 			where logs.seq in (_outdated_logs.seq);
 
 			exit when not found;
+
+			commit ;
 		end loop;
+
+		alter table logs
+		alter column memento set not null;
 	end
 $$;
 
-alter table logs
-alter column memento set not null;

internal/storage/driver/driver.go

@@ -33,37 +33,6 @@ type Driver struct {
 	meter  metric.Meter
 }
 
-func (d *Driver) createLedgerStore(ctx context.Context, db bun.IDB, l *ledger.Ledger) (*ledgerstore.Store, error) {
-
-	b := bucket.New(d.db, l.Bucket)
-	if err := b.Migrate(ctx, d.tracer); err != nil {
-		return nil, fmt.Errorf("migrating bucket: %w", err)
-	}
-
-	_, err := db.NewInsert().
-		Model(l).
-		Returning("id, added_at").
-		Exec(ctx)
-	if err != nil {
-		if errors.Is(postgres.ResolveError(err), postgres.ErrConstraintsFailed{}) {
-			return nil, systemcontroller.ErrLedgerAlreadyExists
-		}
-		return nil, postgres.ResolveError(err)
-	}
-
-	if err := b.AddLedger(ctx, *l, d.db); err != nil {
-		return nil, fmt.Errorf("adding ledger to bucket: %w", err)
-	}
-
-	return ledgerstore.New(
-		d.db,
-		b,
-		*l,
-		ledgerstore.WithMeter(d.meter),
-		ledgerstore.WithTracer(d.tracer),
-	), nil
-}
-
 func (d *Driver) CreateLedger(ctx context.Context, l *ledger.Ledger) (*ledgerstore.Store, error) {
 
 	if l.Metadata == nil {