From d79023c64c89cc76e02b599e73f47b41417e9359 Mon Sep 17 00:00:00 2001
From: Renan Santos <renan.061@gmail.com>
Date: Tue, 27 Aug 2024 14:13:06 -0300
Subject: [PATCH] feat(advancer): add the advancer's repository

---
 .github/workflows/build.yml             |  94 +++++-----
 build/Dockerfile                        |  36 ++++
 internal/node/advancer/advancer.go      |  16 +-
 internal/node/advancer/advancer_test.go |  43 ++---
 internal/repository/advancer.go         | 221 ++++++++++++++++++++++++
 internal/repository/advancer_test.go    | 163 +++++++++++++++++
 internal/repository/base.go             |  17 +-
 internal/repository/base_test.go        |  29 +---
 test/advancer/advancer_test.go          | 206 ++++++++++++++++++++++
 test/tooling/db/db.go                   |  77 +++++++++
 10 files changed, 789 insertions(+), 113 deletions(-)
 create mode 100644 internal/repository/advancer.go
 create mode 100644 internal/repository/advancer_test.go
 create mode 100644 test/advancer/advancer_test.go
 create mode 100644 test/tooling/db/db.go

diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index a92963a60..013e94ea3 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -19,8 +19,57 @@ permissions:
   contents: write
 
 jobs:
+  build-ci-base:
+    runs-on: ubuntu-22.04
+    outputs:
+      output: ${{ steps.export_tag.outputs.image_tag }}
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          submodules: recursive
+
+      - name: Docker meta
+        id: docker_meta
+        uses: docker/metadata-action@v5
+        with:
+          images: |
+            name=ghcr.io/cartesi/rollups-node-ci
+          tags: |
+            type=semver,pattern={{version}}
+            type=ref,event=branch
+            type=ref,event=pr
+
+      - name: Login to GitHub Container Registry
+        uses: docker/login-action@v3
+        with:
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - uses: depot/setup-action@v1
+      - name: Build and push docker image
+        id: docker_build
+        uses: depot/bake-action@v1
+        with:
+          files: |
+            ./docker-bake.hcl
+            ${{ steps.docker_meta.outputs.bake-file }}
+            ./docker-bake.platforms.hcl
+          targets: rollups-node-ci
+          push: true
+          project: ${{ vars.DEPOT_PROJECT }}
+          workdir: build
+
+      - name: Export Image Tag
+        id : export_tag
+        run : echo "image_tag=${{steps.docker_meta.outputs.version}}" >> "$GITHUB_OUTPUT"
+
   do-basic-checks:
     runs-on: ubuntu-22.04
+    container:
+      image: ghcr.io/cartesi/rollups-node-ci:${{needs.build-ci-base.outputs.output}}
+    needs:
+      - build-ci-base
     steps:
       - uses: actions/checkout@v4
         with:
@@ -126,51 +175,6 @@ jobs:
       - name: Run tests
         run: cargo test
 
-  build-ci-base:
-    runs-on: ubuntu-22.04
-    outputs:
-      output: ${{ steps.export_tag.outputs.image_tag }}
-    steps:
-      - uses: actions/checkout@v4
-        with:
-          submodules: recursive
-
-      - name: Docker meta
-        id: docker_meta
-        uses: docker/metadata-action@v5
-        with:
-          images: |
-            name=ghcr.io/cartesi/rollups-node-ci
-          tags: |
-            type=semver,pattern={{version}}
-            type=ref,event=branch
-            type=ref,event=pr
-
-      - name: Login to GitHub Container Registry
-        uses: docker/login-action@v3
-        with:
-          registry: ghcr.io
-          username: ${{ github.actor }}
-          password: ${{ secrets.GITHUB_TOKEN }}
-
-      - uses: depot/setup-action@v1
-      - name: Build and push docker image
-        id: docker_build
-        uses: depot/bake-action@v1
-        with:
-          files: |
-            ./docker-bake.hcl
-            ${{ steps.docker_meta.outputs.bake-file }}
-            ./docker-bake.platforms.hcl
-          targets: rollups-node-ci
-          push: true
-          project: ${{ vars.DEPOT_PROJECT }}
-          workdir: build
-
-      - name: Export Image Tag
-        id : export_tag
-        run : echo "image_tag=${{steps.docker_meta.outputs.version}}" >> "$GITHUB_OUTPUT"
-
   test-go:
     runs-on: ubuntu-22.04
     container:
diff --git a/build/Dockerfile b/build/Dockerfile
index 1187e8c3a..6afa92081 100644
--- a/build/Dockerfile
+++ b/build/Dockerfile
@@ -375,6 +375,42 @@ ARG GO_BUILD_PATH
 ARG ROLLUPS_NODE_VERSION
 WORKDIR ${GO_BUILD_PATH}
 
+ARG MACHINE_EMULATOR_VERSION
+ARG DEBIAN_FRONTEND=noninteractive
+
+# Install ca-certificates and curl (setup).
+RUN <<EOF
+    set -e
+    apt-get update
+    apt-get install -y --no-install-recommends ca-certificates curl
+EOF
+
+# Install the cartesi-machine (cartesi/machine-emulator).
+RUN <<EOF
+    set -e
+    URL=https://github.com/cartesi/machine-emulator/releases/download
+    VERSION=v${MACHINE_EMULATOR_VERSION}
+    ARCH=$(dpkg --print-architecture)
+    ARTIFACT=cartesi-machine-${VERSION}_${ARCH}.deb
+    curl -fsSL ${URL}/${VERSION}/${ARTIFACT} -o ./cartesi-machine.deb
+    apt-get install -y ./cartesi-machine.deb
+    rm ./cartesi-machine.deb
+EOF
+
+# Configure cartesi user and cartesi group.
+RUN <<EOF
+    set -e
+    addgroup --system --gid 102 cartesi
+    adduser --system --uid 102 \
+        --disabled-login \
+        --gecos "cartesi user" \
+        --home /nonexistent \
+        --ingroup cartesi \
+        --no-create-home \
+        --shell /bin/false \
+        cartesi
+EOF
+
 # Download external dependencies.
 COPY go.mod .
 COPY go.sum .
diff --git a/internal/node/advancer/advancer.go b/internal/node/advancer/advancer.go
index 32ef675ab..5e13cf756 100644
--- a/internal/node/advancer/advancer.go
+++ b/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.
diff --git a/internal/node/advancer/advancer_test.go b/internal/node/advancer/advancer_test.go
index 07b6bb6c0..bb97a8d6d 100644
--- a/internal/node/advancer/advancer_test.go
+++ b/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)},
@@ -94,7 +94,9 @@ func (s *AdvancerSuite) TestRun() {
 		require.Len(repository.StoredResults, 3)
 	})
 
-	// NOTE: missing more test cases
+	s.Run("Error/UpdateEpochs", func() {
+		s.T().Skip("TODO")
+	})
 }
 
 func (s *AdvancerSuite) TestProcess() {
@@ -124,7 +126,6 @@ func (s *AdvancerSuite) TestProcess() {
 		err := advancer.process(context.Background(), app, inputs)
 		require.Nil(err)
 		require.Len(repository.StoredResults, 7)
-		require.Equal(*inputs[6], repository.LastInput)
 	})
 
 	s.Run("Panic", func() {
@@ -183,25 +184,7 @@ func (s *AdvancerSuite) TestProcess() {
 			require.Errorf(err, "store-advance error")
 			require.Len(repository.StoredResults, 1)
 		})
-
-		s.Run("UpdateEpochs", func() {
-			require := s.Require()
-
-			_, repository, advancer, app := setup()
-			inputs := []*Input{
-				{Id: 1, RawData: marshal(randomAdvanceResult())},
-				{Id: 2, RawData: marshal(randomAdvanceResult())},
-				{Id: 3, RawData: marshal(randomAdvanceResult())},
-				{Id: 4, RawData: marshal(randomAdvanceResult())},
-			}
-			repository.UpdateEpochsError = errors.New("update-epochs error")
-
-			err := advancer.process(context.Background(), app, inputs)
-			require.Errorf(err, "update-epochs error")
-			require.Len(repository.StoredResults, 4)
-		})
 	})
-
 }
 
 func (s *AdvancerSuite) TestKeysFrom() {
@@ -228,20 +211,19 @@ 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
 }
 
 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 +235,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
 }
 
diff --git a/internal/repository/advancer.go b/internal/repository/advancer.go
new file mode 100644
index 000000000..d0492be9b
--- /dev/null
+++ b/internal/repository/advancer.go
@@ -0,0 +1,221 @@
+// (c) Cartesi and individual authors (see AUTHORS)
+// SPDX-License-Identifier: Apache-2.0 (see LICENSE)
+
+package repository
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+
+	. "github.com/cartesi/rollups-node/internal/node/model"
+	"github.com/cartesi/rollups-node/internal/nodemachine"
+	"github.com/jackc/pgx/v5"
+)
+
+var ErrAdvancerRepository = errors.New("advancer repository error")
+
+type AdvancerRepository struct{ *Database }
+
+func (repo *AdvancerRepository) GetUnprocessedInputs(
+	ctx context.Context,
+	apps []Address,
+) (map[Address][]*Input, error) {
+	result := map[Address][]*Input{}
+	if len(apps) == 0 {
+		return result, nil
+	}
+
+	query := fmt.Sprintf(`
+        SELECT   id, application_address, raw_data
+        FROM     input
+        WHERE    status = 'NONE'
+        AND      application_address IN %s 
+        ORDER BY index ASC, application_address
+    `, toSqlIn(apps)) // NOTE: not sanitized
+	rows, err := repo.db.Query(ctx, query)
+	if err != nil {
+		return nil, fmt.Errorf("%w (failed querying inputs): %w", ErrAdvancerRepository, err)
+	}
+
+	var input Input
+	scans := []any{&input.Id, &input.AppAddress, &input.RawData}
+	_, err = pgx.ForEachRow(rows, scans, func() error {
+		input := input
+		if _, ok := result[input.AppAddress]; ok { //nolint:gosimple
+			result[input.AppAddress] = append(result[input.AppAddress], &input)
+		} else {
+			result[input.AppAddress] = []*Input{&input}
+		}
+		return nil
+	})
+	if err != nil {
+		return nil, fmt.Errorf("%w (failed reading input rows): %w", ErrAdvancerRepository, err)
+	}
+
+	return result, nil
+}
+
+func (repo *AdvancerRepository) StoreAdvanceResult(
+	ctx context.Context,
+	input *Input,
+	res *nodemachine.AdvanceResult,
+) error {
+	tx, err := repo.db.Begin(ctx)
+	if err != nil {
+		return errors.Join(ErrBeginTx, err)
+	}
+
+	// Inserts the outputs.
+	nextOutputIndex, err := repo.getNextIndex(ctx, tx, "output", input.AppAddress)
+	if err != nil {
+		return err
+	}
+	err = repo.insert(ctx, tx, "output", res.Outputs, input.Id, nextOutputIndex)
+	if err != nil {
+		return err
+	}
+
+	// Inserts the reports.
+	nextReportIndex, err := repo.getNextIndex(ctx, tx, "report", input.AppAddress)
+	if err != nil {
+		return err
+	}
+	err = repo.insert(ctx, tx, "report", res.Reports, input.Id, nextReportIndex)
+	if err != nil {
+		return err
+	}
+
+	// Updates the input's status.
+	err = repo.updateInput(ctx, tx, input.Id, res.Status, res.OutputsHash, res.MachineHash)
+	if err != nil {
+		return err
+	}
+
+	err = tx.Commit(ctx)
+	if err != nil {
+		return errors.Join(ErrCommitTx, err, tx.Rollback(ctx))
+	}
+
+	return nil
+}
+
+func (repo *AdvancerRepository) UpdateEpochs(ctx context.Context, app Address) error {
+	query := `
+        UPDATE epoch
+        SET    status = 'PROCESSED_ALL_INPUTS'
+        WHERE  id IN ((
+            SELECT DISTINCT epoch.id
+            FROM   epoch INNER JOIN input ON (epoch.id = input.epoch_id)
+            WHERE  epoch.application_address = @applicationAddress
+            AND    epoch.status = 'CLOSED'
+            AND    input.status != 'NONE'
+        ) EXCEPT (
+            SELECT DISTINCT epoch.id
+            FROM   epoch INNER JOIN input ON (epoch.id = input.epoch_id)
+            WHERE  epoch.application_address = @applicationAddress
+            AND    epoch.status = 'CLOSED'
+            AND    input.status = 'NONE'))
+    `
+	args := pgx.NamedArgs{"applicationAddress": app}
+	_, err := repo.db.Exec(ctx, query, args)
+	if err != nil {
+		return errors.Join(ErrUpdateRow, err)
+	}
+	return nil
+}
+
+// ------------------------------------------------------------------------------------------------
+
+func (_ *AdvancerRepository) getNextIndex(
+	ctx context.Context,
+	tx pgx.Tx,
+	tableName string,
+	appAddress Address,
+) (uint64, error) {
+	var nextIndex uint64
+	query := fmt.Sprintf(`
+        SELECT COALESCE(MAX(%s.index) + 1, 0)
+        FROM   input INNER JOIN %s ON input.id = %s.input_id
+        WHERE  input.status = 'ACCEPTED'
+        AND    input.application_address = $1
+	`, tableName, tableName, tableName)
+	err := tx.QueryRow(ctx, query, appAddress).Scan(&nextIndex)
+	if err != nil {
+		err = fmt.Errorf("failed to get the next %s index: %w", tableName, err)
+		return 0, errors.Join(err, tx.Rollback(ctx))
+	}
+	return nextIndex, nil
+}
+
+func (_ *AdvancerRepository) insert(
+	ctx context.Context,
+	tx pgx.Tx,
+	tableName string,
+	dataArray [][]byte,
+	inputId uint64,
+	nextIndex uint64,
+) error {
+	lenOutputs := int64(len(dataArray))
+	if lenOutputs < 1 {
+		return nil
+	}
+
+	rows := [][]any{}
+	for i, data := range dataArray {
+		rows = append(rows, []any{inputId, nextIndex + uint64(i), data})
+	}
+
+	count, err := tx.CopyFrom(
+		ctx,
+		pgx.Identifier{tableName},
+		[]string{"input_id", "index", "raw_data"},
+		pgx.CopyFromRows(rows),
+	)
+	if err != nil {
+		return errors.Join(ErrCopyFrom, err, tx.Rollback(ctx))
+	}
+	if lenOutputs != count {
+		err := fmt.Errorf("not all %ss were inserted (%d != %d)", tableName, lenOutputs, count)
+		return errors.Join(err, tx.Rollback(ctx))
+	}
+
+	return nil
+}
+
+func (_ *AdvancerRepository) updateInput(
+	ctx context.Context,
+	tx pgx.Tx,
+	inputId uint64,
+	status InputCompletionStatus,
+	outputsHash Hash,
+	machineHash *Hash,
+) error {
+	query := `
+        UPDATE input
+        SET    (status, outputs_hash, machine_hash) = (@status, @outputsHash, @machineHash)
+        WHERE  id = @id
+    `
+	args := pgx.NamedArgs{
+		"status":      status,
+		"outputsHash": outputsHash,
+		"machineHash": machineHash,
+		"id":          inputId,
+	}
+	_, err := tx.Exec(ctx, query, args)
+	if err != nil {
+		return errors.Join(ErrUpdateRow, err, tx.Rollback(ctx))
+	}
+	return nil
+}
+
+// ------------------------------------------------------------------------------------------------
+
+func toSqlIn[T fmt.Stringer](a []T) string {
+	s := []string{}
+	for _, x := range a {
+		s = append(s, fmt.Sprintf("'\\x%s'", x.String()[2:]))
+	}
+	return fmt.Sprintf("(%s)", strings.Join(s, ", "))
+}
diff --git a/internal/repository/advancer_test.go b/internal/repository/advancer_test.go
new file mode 100644
index 000000000..6d9191c98
--- /dev/null
+++ b/internal/repository/advancer_test.go
@@ -0,0 +1,163 @@
+// (c) Cartesi and individual authors (see AUTHORS)
+// SPDX-License-Identifier: Apache-2.0 (see LICENSE)
+
+package repository
+
+import (
+	"context"
+	"testing"
+
+	. "github.com/cartesi/rollups-node/internal/node/model"
+
+	"github.com/cartesi/rollups-node/pkg/rollupsmachine"
+	"github.com/cartesi/rollups-node/test/tooling/db"
+	"github.com/ethereum/go-ethereum/common"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestAdvancerRepository(t *testing.T) {
+	ctx := context.Background()
+
+	endpoint, err := db.Setup(ctx)
+	require.Nil(t, err)
+
+	database, err := Connect(ctx, endpoint)
+	require.Nil(t, err)
+	require.NotNil(t, database)
+
+	app, _, _, err := populate(database)
+	require.Nil(t, err)
+
+	repository := &AdvancerRepository{Database: database}
+
+	t.Run("GetUnprocessedInputs", func(t *testing.T) {
+		t.Skip("TODO")
+	})
+
+	t.Run("StoreAdvanceResult", func(t *testing.T) {
+		t.Skip("TODO")
+	})
+
+	t.Run("UpdateEpochs", func(t *testing.T) {
+		require := require.New(t)
+
+		err = repository.UpdateEpochs(ctx, app.ContractAddress)
+		require.Nil(err)
+
+		epoch0, err := repository.GetEpoch(ctx, 0, app.ContractAddress)
+		require.Nil(err)
+		require.NotNil(epoch0)
+
+		epoch1, err := repository.GetEpoch(ctx, 1, app.ContractAddress)
+		require.Nil(err)
+		require.NotNil(epoch1)
+
+		epoch2, err := repository.GetEpoch(ctx, 2, app.ContractAddress)
+		require.Nil(err)
+		require.NotNil(epoch2)
+
+		epoch3, err := repository.GetEpoch(ctx, 3, app.ContractAddress)
+		require.Nil(err)
+		require.NotNil(epoch3)
+
+		require.Equal(EpochStatusProcessedAllInputs, epoch0.Status)
+		require.Equal(EpochStatusProcessedAllInputs, epoch1.Status)
+		require.Equal(EpochStatusClosed, epoch2.Status)
+		require.Equal(EpochStatusOpen, epoch3.Status)
+	})
+}
+
+// ------------------------------------------------------------------------------------------------
+
+func populate(database *Database) (*Application, []*Epoch, []*Input, error) {
+	ctx := context.Background()
+
+	app := &Application{
+		ContractAddress:    common.HexToAddress("deadbeef"),
+		IConsensusAddress:  common.HexToAddress("beefdead"),
+		TemplateHash:       [32]byte{},
+		LastProcessedBlock: 0,
+		Status:             "RUNNING",
+	}
+
+	err := database.InsertApplication(ctx, app)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+
+	epochs := []*Epoch{{
+		FirstBlock: 0,
+		LastBlock:  1,
+		Status:     EpochStatusClosed,
+	}, {
+		FirstBlock: 2,
+		LastBlock:  3,
+		Status:     EpochStatusClosed,
+	}, {
+		FirstBlock: 4,
+		LastBlock:  5,
+		Status:     EpochStatusClosed,
+	}, {
+		FirstBlock: 6,
+		LastBlock:  7,
+		Status:     EpochStatusOpen,
+	}}
+
+	for i, epoch := range epochs {
+		epoch.Index = uint64(i)
+		epoch.AppAddress = app.ContractAddress
+		epoch.Id, err = database.InsertEpoch(ctx, epoch)
+		if err != nil {
+			return nil, nil, nil, err
+		}
+	}
+
+	inputs := []*Input{{
+		EpochId:          epochs[0].Id,
+		CompletionStatus: InputStatusAccepted,
+		RawData:          []byte("first input"),
+	}, {
+		EpochId:          epochs[0].Id,
+		CompletionStatus: InputStatusRejected,
+		RawData:          []byte("second input"),
+	}, {
+		EpochId:          epochs[1].Id,
+		CompletionStatus: InputStatusException,
+		RawData:          []byte("third input"),
+	}, {
+		EpochId:          epochs[1].Id,
+		CompletionStatus: InputStatusAccepted,
+		RawData:          []byte("fourth input"),
+	}, {
+		EpochId:          epochs[2].Id,
+		CompletionStatus: InputStatusAccepted,
+		RawData:          []byte("fifth input"),
+	}, {
+		EpochId:          epochs[2].Id,
+		CompletionStatus: InputStatusNone,
+		RawData:          []byte("sixth input"),
+	}, {
+		EpochId:          epochs[3].Id,
+		CompletionStatus: InputStatusNone,
+		RawData:          []byte("seventh input"),
+	}}
+
+	for i, input := range inputs {
+		input.Index = uint64(i)
+		input.BlockNumber = uint64(i)
+		input.AppAddress = app.ContractAddress
+
+		input.RawData, err = rollupsmachine.Input{Data: input.RawData}.Encode()
+		if err != nil {
+			return nil, nil, nil, err
+		}
+
+		input.Id, err = database.InsertInput(ctx, input)
+		if err != nil {
+			return nil, nil, nil, err
+		}
+	}
+
+	return app, epochs, inputs, nil
+}
diff --git a/internal/repository/base.go b/internal/repository/base.go
index 10f688bc5..3cc6cf4d0 100644
--- a/internal/repository/base.go
+++ b/internal/repository/base.go
@@ -19,7 +19,14 @@ type Database struct {
 	db *pgxpool.Pool
 }
 
-var ErrInsertRow = errors.New("unable to insert row")
+var (
+	ErrInsertRow = errors.New("unable to insert row")
+	ErrUpdateRow = errors.New("unable to update row")
+	ErrCopyFrom  = errors.New("unable to COPY FROM")
+
+	ErrBeginTx  = errors.New("unable to begin transaction")
+	ErrCommitTx = errors.New("unable to commit transaction")
+)
 
 func Connect(
 	ctx context.Context,
@@ -119,7 +126,6 @@ func (pg *Database) InsertEpoch(
 	ctx context.Context,
 	epoch *Epoch,
 ) (uint64, error) {
-
 	var id uint64
 
 	query := `
@@ -140,7 +146,8 @@ func (pg *Database) InsertEpoch(
 		@status,
 		@applicationAddress)
 	RETURNING
-		id`
+		id
+    `
 
 	args := pgx.NamedArgs{
 		"index":              epoch.Index,
@@ -283,7 +290,9 @@ func (pg *Database) InsertSnapshot(
 		(@inputId,
 		@appAddress,
 		@uri)
-	RETURNING id`
+	RETURNING
+        id
+    `
 
 	args := pgx.NamedArgs{
 		"inputId":    snapshot.InputId,
diff --git a/internal/repository/base_test.go b/internal/repository/base_test.go
index be5cd122f..1abd680ab 100644
--- a/internal/repository/base_test.go
+++ b/internal/repository/base_test.go
@@ -10,6 +10,8 @@ import (
 	"time"
 
 	. "github.com/cartesi/rollups-node/internal/node/model"
+	"github.com/cartesi/rollups-node/internal/repository/schema"
+
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/stretchr/testify/suite"
 	"github.com/testcontainers/testcontainers-go"
@@ -38,10 +40,10 @@ func (s *RepositorySuite) SetupSuite() {
 	endpoint, err := s.postgres.ConnectionString(s.ctx, "sslmode=disable")
 	s.Require().Nil(err)
 
-	schemaManager, err := NewSchemaManager(endpoint)
+	schema, err := schema.New(endpoint)
 	s.Require().Nil(err)
 
-	err = schemaManager.Upgrade()
+	err = schema.Upgrade()
 	s.Require().Nil(err)
 
 	s.database, err = Connect(s.ctx, endpoint)
@@ -334,18 +336,6 @@ func (s *RepositorySuite) TestOutputDoesntExist() {
 	s.Require().Nil(err)
 }
 
-func (s *RepositorySuite) TestOutputFailsDuplicateRow() {
-	output := Output{
-		Index:                1,
-		InputId:              1,
-		RawData:              common.Hex2Bytes("deadbeef"),
-		OutputHashesSiblings: nil,
-	}
-
-	_, err := s.database.InsertOutput(s.ctx, &output)
-	s.Require().ErrorContains(err, "duplicate key value")
-}
-
 func (s *RepositorySuite) TestOutputFailsInputDoesntExist() {
 	output := Output{
 		Index:                10,
@@ -377,17 +367,6 @@ func (s *RepositorySuite) TestReportDoesntExist() {
 	s.Require().Nil(err)
 }
 
-func (s *RepositorySuite) TestReportFailsDuplicateRow() {
-	report := Report{
-		Index:   1,
-		InputId: 1,
-		RawData: common.Hex2Bytes("deadbeef"),
-	}
-
-	err := s.database.InsertReport(s.ctx, &report)
-	s.Require().ErrorContains(err, "duplicate key value")
-}
-
 func (s *RepositorySuite) TestReportFailsInputDoesntExist() {
 	report := Report{
 		Index:   2,
diff --git a/test/advancer/advancer_test.go b/test/advancer/advancer_test.go
new file mode 100644
index 000000000..aeeae06ff
--- /dev/null
+++ b/test/advancer/advancer_test.go
@@ -0,0 +1,206 @@
+// (c) Cartesi and individual authors (see AUTHORS)
+// SPDX-License-Identifier: Apache-2.0 (see LICENSE)
+
+package advancer
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	. "github.com/cartesi/rollups-node/internal/node/model"
+
+	"github.com/cartesi/rollups-node/internal/node/advancer"
+	"github.com/cartesi/rollups-node/internal/nodemachine"
+	"github.com/cartesi/rollups-node/internal/repository"
+	"github.com/cartesi/rollups-node/pkg/emulator"
+	"github.com/cartesi/rollups-node/pkg/rollupsmachine"
+	"github.com/cartesi/rollups-node/pkg/rollupsmachine/cartesimachine"
+	"github.com/cartesi/rollups-node/test/snapshot"
+	"github.com/cartesi/rollups-node/test/tooling/db"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/stretchr/testify/require"
+)
+
+func TestAdvancer(t *testing.T) {
+	require := require.New(t)
+	ctx := context.Background()
+
+	// Setups the database.
+	endpoint, err := db.Setup(ctx)
+	require.Nil(err)
+	database, err := repository.Connect(ctx, endpoint)
+	require.Nil(err)
+	require.NotNil(database)
+	app, _, _, err := populate(database)
+	require.Nil(err)
+
+	// Creates the snapshot.
+	script := "ioctl-echo-loop --vouchers=1 --notices=3 --reports=5 --verbose=1"
+	snapshot, err := snapshot.FromScript(script, uint64(1_000_000_000))
+	require.Nil(err)
+	defer func() { require.Nil(snapshot.Close()) }()
+
+	// Starts the server.
+	verbosity := cartesimachine.ServerVerbosityInfo
+	address, err := cartesimachine.StartServer(verbosity, 0, os.Stdout, os.Stderr)
+	require.Nil(err)
+
+	// Loads the cartesimachine.
+	config := &emulator.MachineRuntimeConfig{}
+	cartesiMachine, err := cartesimachine.Load(ctx, snapshot.Path(), address, config)
+	require.Nil(err)
+	require.NotNil(cartesiMachine)
+
+	// Wraps the cartesimachine with rollupsmachine.
+	rollupsMachine, err := rollupsmachine.New(ctx, cartesiMachine, 50_000_000, 5_000_000_000)
+	require.Nil(err)
+	require.NotNil(rollupsMachine)
+
+	// Wraps the rollupsmachine with nodemachine.
+	nodeMachine, err := nodemachine.New(rollupsMachine, 0, time.Minute, time.Minute, 10)
+	require.Nil(err)
+	require.NotNil(nodeMachine)
+	defer func() { require.Nil(nodeMachine.Close()) }()
+
+	// Creates the machine pool.
+	machines := advancer.Machines{app.ContractAddress: nodeMachine}
+
+	// Creates the advancer's repository.
+	repository := &repository.AdvancerRepository{Database: database}
+
+	// Creates the advancer.
+	advancer, err := advancer.New(machines, repository)
+	require.Nil(err)
+	require.NotNil(advancer)
+
+	// Creates the poller from the advancer.
+	poller, err := advancer.Poller(5 * time.Second)
+	require.Nil(err)
+	require.NotNil(poller)
+
+	// Starts the advancer in a separate goroutine.
+	done := make(chan struct{}, 1)
+	go func() {
+		ready := make(chan struct{}, 1)
+		err = poller.Start(ctx, ready)
+		<-ready
+		require.Nil(err, "%v", err)
+		done <- struct{}{}
+	}()
+
+	// Orders the advancer to stop after some time has passed.
+	time.Sleep(5 * time.Second)
+	poller.Stop()
+
+wait:
+	for {
+		select {
+		case <-done:
+			fmt.Println("Done!")
+			break wait
+		default:
+			fmt.Println("Waiting...")
+			time.Sleep(time.Second)
+		}
+	}
+
+	t.Run("AssertThings!", func(t *testing.T) {
+		t.Skip("TODO")
+	})
+}
+
+func populate(database *repository.Database) (*Application, []*Epoch, []*Input, error) {
+	ctx := context.Background()
+
+	app := &Application{
+		ContractAddress:    common.HexToAddress("deadbeef"),
+		IConsensusAddress:  common.HexToAddress("beefdead"),
+		TemplateHash:       [32]byte{},
+		LastProcessedBlock: 0,
+		Status:             "RUNNING",
+	}
+
+	err := database.InsertApplication(ctx, app)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+
+	epochs := []*Epoch{{
+		FirstBlock: 0,
+		LastBlock:  1,
+		Status:     EpochStatusClosed,
+	}, {
+		FirstBlock: 2,
+		LastBlock:  3,
+		Status:     EpochStatusClosed,
+	}, {
+		FirstBlock: 4,
+		LastBlock:  5,
+		Status:     EpochStatusClosed,
+	}, {
+		FirstBlock: 6,
+		LastBlock:  7,
+		Status:     EpochStatusOpen,
+	}}
+
+	for i, epoch := range epochs {
+		epoch.Index = uint64(i)
+		epoch.AppAddress = app.ContractAddress
+		epoch.Id, err = database.InsertEpoch(ctx, epoch)
+		if err != nil {
+			return nil, nil, nil, err
+		}
+	}
+
+	inputs := []*Input{{
+		EpochId:          epochs[0].Id,
+		CompletionStatus: InputStatusAccepted,
+		RawData:          []byte("first input"),
+	}, {
+		EpochId:          epochs[0].Id,
+		CompletionStatus: InputStatusRejected,
+		RawData:          []byte("second input"),
+	}, {
+		EpochId:          epochs[1].Id,
+		CompletionStatus: InputStatusException,
+		RawData:          []byte("third input"),
+	}, {
+		EpochId:          epochs[1].Id,
+		CompletionStatus: InputStatusAccepted,
+		RawData:          []byte("fourth input"),
+	}, {
+		EpochId:          epochs[2].Id,
+		CompletionStatus: InputStatusAccepted,
+		RawData:          []byte("fifth input"),
+	}, {
+		EpochId:          epochs[2].Id,
+		CompletionStatus: InputStatusNone,
+		RawData:          []byte("sixth input"),
+	}, {
+		EpochId:          epochs[3].Id,
+		CompletionStatus: InputStatusNone,
+		RawData:          []byte("seventh input"),
+	}}
+
+	for i, input := range inputs {
+		input.Index = uint64(i)
+		input.BlockNumber = uint64(i)
+		input.AppAddress = app.ContractAddress
+
+		input.RawData, err = rollupsmachine.Input{Data: input.RawData}.Encode()
+		if err != nil {
+			return nil, nil, nil, err
+		}
+
+		input.Id, err = database.InsertInput(ctx, input)
+		if err != nil {
+			return nil, nil, nil, err
+		}
+	}
+
+	return app, epochs, inputs, nil
+}
diff --git a/test/tooling/db/db.go b/test/tooling/db/db.go
new file mode 100644
index 000000000..2520bc83b
--- /dev/null
+++ b/test/tooling/db/db.go
@@ -0,0 +1,77 @@
+// (c) Cartesi and individual authors (see AUTHORS)
+// SPDX-License-Identifier: Apache-2.0 (see LICENSE)
+
+package db
+
+import (
+	"context"
+	"os"
+	"time"
+
+	"github.com/cartesi/rollups-node/internal/repository/schema"
+	"github.com/testcontainers/testcontainers-go"
+	"github.com/testcontainers/testcontainers-go/modules/postgres"
+	"github.com/testcontainers/testcontainers-go/wait"
+)
+
+const (
+	postgresImage    = "postgres:16-alpine"
+	postgresDatabase = "cartesinode"
+	postgresUsername = "admin"
+	postgresPassword = "password"
+)
+
+func Setup(ctx context.Context) (string, error) {
+	endpoint, ok := os.LookupEnv("TEST_POSTGRES_ENDPOINT")
+	if !ok {
+		container, err := SetupContainer(ctx)
+		if err != nil {
+			return "", err
+		}
+
+		endpoint, err = container.ConnectionString(ctx, "sslmode=disable")
+		if err != nil {
+			return "", err
+		}
+	}
+
+	err := SetupSchema(endpoint)
+	if err != nil {
+		return "", err
+	}
+
+	return endpoint, nil
+}
+
+func SetupContainer(ctx context.Context) (*postgres.PostgresContainer, error) {
+	log := "database system is ready to accept connections"
+	occurrences := 2            //nolint: mnd
+	timeout := 10 * time.Second //nolint: mnd
+	strategy := wait.ForLog(log).WithOccurrence(occurrences).WithStartupTimeout(timeout)
+	return postgres.Run(ctx,
+		postgresImage,
+		postgres.WithDatabase(postgresDatabase),
+		postgres.WithUsername(postgresUsername),
+		postgres.WithPassword(postgresPassword),
+		testcontainers.WithWaitStrategy(strategy))
+}
+
+func SetupSchema(endpoint string) error {
+	schema, err := schema.New(endpoint)
+	if err != nil {
+		return err
+	}
+	defer schema.Close()
+
+	err = schema.Downgrade()
+	if err != nil {
+		return err
+	}
+
+	err = schema.Upgrade()
+	if err != nil {
+		return err
+	}
+
+	return nil
+}