insert_state_query_test.go

package pgdriver

import (
	"context"
	"errors"
	"fmt"
	"testing"

	"github.com/jackc/pgx/v5/pgconn"
	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/makasim/flowstate"
	"github.com/makasim/flowstate/pgdriver/testpgdriver"
	"github.com/stretchr/testify/require"
)

func TestQuery_InsertState(main *testing.T) {
	openDB := func(t *testing.T, dsn0, dbName string) *pgxpool.Pool {
		conn := testpgdriver.OpenFreshDB(t, dsn0, dbName)

		for i, m := range Migrations {
			_, err := conn.Exec(context.Background(), m.SQL)
			require.NoError(t, err, fmt.Sprintf("Migration #%d (%s) failed ", i, m.Desc))
		}

		return conn
	}

	main.Run("IDEmpty", func(t *testing.T) {
		conn := openDB(t, `postgres://postgres:postgres@localhost:5432/postgres`, ``)

		q := &queries{}

		s := flowstate.State{ID: ``}
		err := q.InsertState(context.Background(), conn, &s)
		require.EqualError(t, err, `id is empty`)
	})

	main.Run("RevNotEmpty", func(t *testing.T) {
		conn := openDB(t, `postgres://postgres:postgres@localhost:5432/postgres`, ``)

		q := &queries{}

		s := flowstate.State{ID: `anID`, Rev: 1}
		err := q.InsertState(context.Background(), conn, &s)
		require.EqualError(t, err, `rev is not empty`)
	})

	main.Run("OK", func(t *testing.T) {
		conn := openDB(t, `postgres://postgres:postgres@localhost:5432/postgres`, ``)

		q := &queries{}

		s := flowstate.State{ID: `anID`}
		err := q.InsertState(context.Background(), conn, &s)
		require.NoError(t, err)
		require.Greater(t, s.Rev, int64(0))

		require.Equal(t, []testpgdriver.StateRow{
			{
				ID:     "anID",
				Rev:    1,
				Labels: nil,
				State: flowstate.State{
					ID:  "anID",
					Rev: 0, // rev is not updated in json
				},
			},
		}, testpgdriver.FindAllStates(t, conn))

		require.Equal(t, []testpgdriver.LatestStateRow{
			{
				ID:  "anID",
				Rev: 1,
			},
		}, testpgdriver.FindAllLatestStates(t, conn))
	})

	main.Run("FullState", func(t *testing.T) {
		conn := openDB(t, `postgres://postgres:postgres@localhost:5432/postgres`, ``)

		q := &queries{}

		s := flowstate.State{
			ID:  `anID`,
			Rev: 0,
			Annotations: map[string]string{
				`fooAnnotKey`: `fooAnnotVal`,
				`barAnnotKey`: `barAnnotVal`,
			},
			Labels: map[string]string{
				`fooLabelKey`: `fooLabelVal`,
				`barLabelKey`: `barLabelVal`,
			},
			CommittedAtUnixMilli: 123,
			Transition: flowstate.Transition{
				FromID: "fromFlowID",
				ToID:   "toFlowID",
				Annotations: map[string]string{
					`fooTsAnnotKey`: `fooTsAnnotVal`,
					`barTsAnnotKey`: `barTsAnnotVal`,
				},
			},
		}
		err := q.InsertState(context.Background(), conn, &s)
		require.NoError(t, err)
		require.Greater(t, s.Rev, int64(0))

		require.Equal(t, []testpgdriver.StateRow{
			{
				ID:  "anID",
				Rev: 1,
				Labels: map[string]string{
					`fooLabelKey`: `fooLabelVal`,
					`barLabelKey`: `barLabelVal`,
				},
				State: flowstate.State{
					ID:  `anID`,
					Rev: 0, // rev is not updated in json
					Annotations: map[string]string{
						`fooAnnotKey`: `fooAnnotVal`,
						`barAnnotKey`: `barAnnotVal`,
					},
					Labels: map[string]string{
						`fooLabelKey`: `fooLabelVal`,
						`barLabelKey`: `barLabelVal`,
					},
					CommittedAtUnixMilli: 123,
					Transition: flowstate.Transition{
						FromID: "fromFlowID",
						ToID:   "toFlowID",
						Annotations: map[string]string{
							`fooTsAnnotKey`: `fooTsAnnotVal`,
							`barTsAnnotKey`: `barTsAnnotVal`,
						},
					},
				},
			},
		}, testpgdriver.FindAllStates(t, conn))
	})

	main.Run("Several", func(t *testing.T) {
		conn := openDB(t, `postgres://postgres:postgres@localhost:5432/postgres`, ``)

		q := &queries{}

		s0 := flowstate.State{
			ID:  `aFooID`,
			Rev: 0,
		}
		err := q.InsertState(context.Background(), conn, &s0)
		require.NoError(t, err)
		require.Greater(t, s0.Rev, int64(0))

		s1 := flowstate.State{
			ID:  `aBarID`,
			Rev: 0,
		}
		err = q.InsertState(context.Background(), conn, &s1)
		require.NoError(t, err)
		require.Greater(t, s1.Rev, int64(0))

		require.Equal(t, []testpgdriver.StateRow{
			{
				ID:     "aBarID",
				Rev:    2,
				Labels: nil,
				State: flowstate.State{
					ID:  `aBarID`,
					Rev: 0, // rev is not updated in json
				},
			},
			{
				ID:     "aFooID",
				Rev:    1,
				Labels: nil,
				State: flowstate.State{
					ID:  `aFooID`,
					Rev: 0, // rev is not updated in json
				},
			},
		}, testpgdriver.FindAllStates(t, conn))

		require.Equal(t, []testpgdriver.LatestStateRow{
			{
				ID:  "aBarID",
				Rev: 2,
			},
			{
				ID:  "aFooID",
				Rev: 1,
			},
		}, testpgdriver.FindAllLatestStates(t, conn))
	})

	main.Run("Conflict", func(t *testing.T) {
		conn := openDB(t, `postgres://postgres:postgres@localhost:5432/postgres`, ``)

		q := &queries{}

		s0 := flowstate.State{ID: `anID`}
		err := q.InsertState(context.Background(), conn, &s0)
		require.NoError(t, err)
		require.Greater(t, s0.Rev, int64(0))

		s1 := flowstate.State{ID: `anID`}
		err = q.InsertState(context.Background(), conn, &s1)
		require.EqualError(t, err, `ERROR: duplicate key value violates unique constraint "flowstate_latest_states_pkey" (SQLSTATE 23505)`)
		pgErr := &pgconn.PgError{}
		require.True(t, errors.As(err, &pgErr))
		require.Equal(t, `23505`, pgErr.Code)
		require.Equal(t, `flowstate_latest_states_pkey`, pgErr.ConstraintName)

		require.Equal(t, []testpgdriver.StateRow{
			{
				ID:     "anID",
				Rev:    1,
				Labels: nil,
				State: flowstate.State{
					ID:  "anID",
					Rev: 0, // rev is not updated in json
				},
			},
		}, testpgdriver.FindAllStates(t, conn))

		require.Equal(t, []testpgdriver.LatestStateRow{
			{
				ID:  "anID",
				Rev: 1,
			},
		}, testpgdriver.FindAllLatestStates(t, conn))
	})

	main.Run("TxOK", func(t *testing.T) {
		conn := openDB(t, `postgres://postgres:postgres@localhost:5432/postgres`, ``)

		tx, err := conn.Begin(context.Background())
		require.NoError(t, err)
		defer tx.Rollback(context.Background())

		q := &queries{}

		s := flowstate.State{ID: `anID`}
		err = q.InsertState(context.Background(), tx, &s)
		require.NoError(t, err)
		require.Greater(t, s.Rev, int64(0))

		require.NoError(t, tx.Commit(context.Background()))

		require.Equal(t, []testpgdriver.StateRow{
			{
				ID:     "anID",
				Rev:    1,
				Labels: nil,
				State: flowstate.State{
					ID:  "anID",
					Rev: 0, // rev is not updated in json
				},
			},
		}, testpgdriver.FindAllStates(t, conn))

		require.Equal(t, []testpgdriver.LatestStateRow{
			{
				ID:  "anID",
				Rev: 1,
			},
		}, testpgdriver.FindAllLatestStates(t, conn))
	})

	main.Run("TxConflict", func(t *testing.T) {
		conn := openDB(t, `postgres://postgres:postgres@localhost:5432/postgres`, ``)

		tx, err := conn.Begin(context.Background())
		require.NoError(t, err)
		defer tx.Rollback(context.Background())

		q := &queries{}

		s0 := flowstate.State{ID: `anID`}
		err = q.InsertState(context.Background(), tx, &s0)
		require.NoError(t, err)
		require.Greater(t, s0.Rev, int64(0))

		require.NoError(t, tx.Commit(context.Background()))

		s1 := flowstate.State{ID: `anID`}
		err = q.InsertState(context.Background(), conn, &s1)
		require.EqualError(t, err, `ERROR: duplicate key value violates unique constraint "flowstate_latest_states_pkey" (SQLSTATE 23505)`)
		pgErr := &pgconn.PgError{}
		require.True(t, errors.As(err, &pgErr))
		require.Equal(t, `23505`, pgErr.Code)
		require.Equal(t, `flowstate_latest_states_pkey`, pgErr.ConstraintName)

		require.Equal(t, []testpgdriver.StateRow{
			{
				ID:     "anID",
				Rev:    1,
				Labels: nil,
				State: flowstate.State{
					ID:  "anID",
					Rev: 0, // rev is not updated in json
				},
			},
		}, testpgdriver.FindAllStates(t, conn))

		require.Equal(t, []testpgdriver.LatestStateRow{
			{
				ID:  "anID",
				Rev: 1,
			},
		}, testpgdriver.FindAllLatestStates(t, conn))
	})
}