feat(advancer): add the advancer's repository

cmd/cartesi-rollups-cli/root/db/check/check.go

@@ -3,10 +3,10 @@
 package check
 
 import (
-	"fmt"
+	"log/slog"
 
 	"github.com/cartesi/rollups-node/cmd/cartesi-rollups-cli/root/common"
-	"github.com/cartesi/rollups-node/internal/repository"
+	"github.com/cartesi/rollups-node/internal/repository/schema"
 	"github.com/spf13/cobra"
 )
 
@@ -17,13 +17,12 @@ var Cmd = &cobra.Command{
 }
 
 func run(cmd *cobra.Command, args []string) {
-
-	schemaManager, err := repository.NewSchemaManager(common.PostgresEndpoint)
+	schema, err := schema.New(common.PostgresEndpoint)
 	cobra.CheckErr(err)
-	defer schemaManager.Close()
+	defer schema.Close()
 
-	err = schemaManager.ValidateSchemaVersion()
+	version, err := schema.ValidateVersion()
 	cobra.CheckErr(err)
 
-	fmt.Printf("Database Schema is at the correct version: %d\n", repository.EXPECTED_VERSION)
+	slog.Info("Database Schema is at the correct version.", "version", version)
 }

cmd/cartesi-rollups-cli/root/db/upgrade/upgrade.go

@@ -3,11 +3,10 @@
 package upgrade
 
 import (
-	"fmt"
 	"log/slog"
 
 	"github.com/cartesi/rollups-node/cmd/cartesi-rollups-cli/root/common"
-	"github.com/cartesi/rollups-node/internal/repository"
+	"github.com/cartesi/rollups-node/internal/repository/schema"
 	"github.com/spf13/cobra"
 )
 
@@ -18,21 +17,15 @@ var Cmd = &cobra.Command{
 }
 
 func run(cmd *cobra.Command, args []string) {
-
-	schemaManager, err := repository.NewSchemaManager(common.PostgresEndpoint)
+	schema, err := schema.New(common.PostgresEndpoint)
 	cobra.CheckErr(err)
-	defer schemaManager.Close()
+	defer schema.Close()
 
-	err = schemaManager.Upgrade()
+	err = schema.Up()
 	cobra.CheckErr(err)
 
-	version, err := schemaManager.GetVersion()
+	version, err := schema.ValidateVersion()
 	cobra.CheckErr(err)
 
-	if repository.EXPECTED_VERSION != version {
-		slog.Warn("Current version is different to expected one")
-	}
-
-	fmt.Printf("Database Schema successfully Updated. Current version is %d\n", version)
-
+	slog.Info("Database Schema successfully Updated.", "version", version)
 }

internal/node/advancer/advancer.go

@@ -67,6 +67,14 @@ func (advancer *Advancer) Step(ctx context.Context) error {
 		}
 	}
 
+	// Updates the status of the epochs.
+	for _, app := range apps {
+		err := advancer.repository.UpdateEpochs(ctx, app)
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 
@@ -99,11 +107,7 @@ func (advancer *Advancer) process(ctx context.Context, app Address, inputs []*In
 		}
 	}
 
-	// Updates the status of the epochs based on the last processed input.
-	lastInput := inputs[len(inputs)-1]
-	err := advancer.repository.UpdateEpochs(ctx, app, lastInput)
-
-	return err
+	return nil
 }
 
 // ------------------------------------------------------------------------------------------------
@@ -114,7 +118,7 @@ type Repository interface {
 
 	StoreAdvanceResult(context.Context, *Input, *nodemachine.AdvanceResult) error
 
-	UpdateEpochs(_ context.Context, app Address, lastInput *Input) error
+	UpdateEpochs(_ context.Context, app Address) error
 }
 
 // A map of application addresses to machines.

internal/node/advancer/advancer_test.go

@@ -73,7 +73,7 @@ func (s *AdvancerSuite) TestRun() {
 		res3 := randomAdvanceResult()
 
 		repository := &MockRepository{
-			GetInputsReturn: map[Address][]*Input{
+			GetUnprocessedInputsReturn: map[Address][]*Input{
 				app1: {
 					{Id: 1, RawData: marshal(res1)},
 					{Id: 2, RawData: marshal(res2)},
@@ -228,10 +228,10 @@ func (mock *MockMachine) Advance(
 // ------------------------------------------------------------------------------------------------
 
 type MockRepository struct {
-	GetInputsReturn   map[Address][]*Input
-	GetInputsError    error
-	StoreAdvanceError error
-	UpdateEpochsError error
+	GetUnprocessedInputsReturn map[Address][]*Input
+	GetUnprocessedInputsError  error
+	StoreAdvanceError          error
+	UpdateEpochsError          error
 
 	StoredResults []*nodemachine.AdvanceResult
 	LastInput     Input
@@ -241,7 +241,7 @@ func (mock *MockRepository) GetUnprocessedInputs(
 	_ context.Context,
 	appAddresses []Address,
 ) (map[Address][]*Input, error) {
-	return mock.GetInputsReturn, mock.GetInputsError
+	return mock.GetUnprocessedInputsReturn, mock.GetUnprocessedInputsError
 }
 
 func (mock *MockRepository) StoreAdvanceResult(
@@ -253,12 +253,7 @@ func (mock *MockRepository) StoreAdvanceResult(
 	return mock.StoreAdvanceError
 }
 
-func (mock *MockRepository) UpdateEpochs(
-	_ context.Context,
-	_ Address,
-	lastInput *Input,
-) error {
-	mock.LastInput = *lastInput
+func (mock *MockRepository) UpdateEpochs(_ context.Context, _ Address) error {
 	return mock.UpdateEpochsError
 }
 

internal/node/startup/startup.go

@@ -12,6 +12,7 @@ import (
 	"github.com/cartesi/rollups-node/internal/node/config"
 	"github.com/cartesi/rollups-node/internal/node/model"
 	"github.com/cartesi/rollups-node/internal/repository"
+	"github.com/cartesi/rollups-node/internal/repository/schema"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/jackc/pgx/v5"
 	"github.com/lmittmann/tint"
@@ -20,31 +21,19 @@ import (
 
 // Validates the Node Database Schema Version
 func ValidateSchema(config config.NodeConfig) error {
-	var (
-		schemaManager *repository.SchemaManager
-		err           error
-	)
-
-	if !config.PostgresSslMode {
-		schemaManager, err = repository.NewSchemaManager(
-			fmt.Sprintf("%v?sslmode=disable", config.PostgresEndpoint.Value))
-		if err != nil {
-			return err
-		}
-	} else {
-		schemaManager, err = repository.NewSchemaManager(config.PostgresEndpoint.Value)
-		if err != nil {
-			return err
-		}
+	endpoint := config.PostgresEndpoint.Value
+	if config.PostgresSslMode {
+		endpoint += "?sslmode=disable"
 	}
-	defer schemaManager.Close()
-	err = schemaManager.ValidateSchemaVersion()
+
+	schema, err := schema.New(endpoint)
 	if err != nil {
 		return err
 	}
+	defer schema.Close()
 
-	return nil
-
+	_, err = schema.ValidateVersion()
+	return err
 }
 
 // Configure the node logs