Repository: xataio/pgstream
Pull request #79 (2024-10-02 11:26:45)

Add snapshot store
This PR adds a snapshot store to persist snapshot requests. This will be used by a snapshot server which will receive the requests, in order to have a persisted list of snapshots and their status.

Part of https://github.com/xataio/pgstream/issues/80


Patch:
diff --git a/internal/postgres/mocks/mock_pg_querier.go b/internal/postgres/mocks/mock_pg_querier.go
index c69c060..36915de 100644
--- a/internal/postgres/mocks/mock_pg_querier.go
+++ b/internal/postgres/mocks/mock_pg_querier.go
@@ -11,9 +11,10 @@ import (
 type Querier struct {
 	QueryRowFn func(ctx context.Context, query string, args ...any) postgres.Row
 	QueryFn    func(ctx context.Context, query string, args ...any) (postgres.Rows, error)
-	ExecFn     func(context.Context, string, ...any) (postgres.CommandTag, error)
+	ExecFn     func(context.Context, uint, string, ...any) (postgres.CommandTag, error)
 	ExecInTxFn func(context.Context, func(tx postgres.Tx) error) error
 	CloseFn    func(context.Context) error
+	execCalls  uint
 }
 
 func (m *Querier) QueryRow(ctx context.Context, query string, args ...any) postgres.Row {
@@ -25,7 +26,8 @@ func (m *Querier) Query(ctx context.Context, query string, args ...any) (postgre
 }
 
 func (m *Querier) Exec(ctx context.Context, query string, args ...any) (postgres.CommandTag, error) {
-	return m.ExecFn(ctx, query, args...)
+	m.execCalls++
+	return m.ExecFn(ctx, m.execCalls, query, args...)
 }
 
 func (m *Querier) ExecInTx(ctx context.Context, fn func(tx postgres.Tx) error) error {
diff --git a/internal/postgres/mocks/mock_rows.go b/internal/postgres/mocks/mock_rows.go
new file mode 100644
index 0000000..213a738
--- /dev/null
+++ b/internal/postgres/mocks/mock_rows.go
@@ -0,0 +1,56 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package mocks
+
+import (
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+)
+
+type Rows struct {
+	CloseFn             func()
+	ErrFn               func() error
+	FieldDescriptionsFn func() []pgconn.FieldDescription
+	NextFn              func(i uint) bool
+	ScanFn              func(dest ...any) error
+	ValuesFn            func() ([]any, error)
+	RawValuesFn         func() [][]byte
+	nextCalls           uint
+}
+
+func (m *Rows) Close() {
+	m.CloseFn()
+}
+
+func (m *Rows) Err() error {
+	return m.ErrFn()
+}
+
+func (m *Rows) CommandTag() pgconn.CommandTag {
+	return pgconn.CommandTag{}
+}
+
+func (m *Rows) FieldDescriptions() []pgconn.FieldDescription {
+	return m.FieldDescriptionsFn()
+}
+
+func (m *Rows) Next() bool {
+	m.nextCalls++
+	return m.NextFn(m.nextCalls)
+}
+
+func (m *Rows) Scan(dest ...any) error {
+	return m.ScanFn(dest...)
+}
+
+func (m *Rows) Values() ([]any, error) {
+	return m.ValuesFn()
+}
+
+func (m *Rows) RawValues() [][]byte {
+	return m.RawValuesFn()
+}
+
+func (m *Rows) Conn() *pgx.Conn {
+	return &pgx.Conn{}
+}
diff --git a/pkg/snapshot/snapshot.go b/pkg/snapshot/snapshot.go
new file mode 100644
index 0000000..f794e72
--- /dev/null
+++ b/pkg/snapshot/snapshot.go
@@ -0,0 +1,32 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package snapshot
+
+type Snapshot struct {
+	SchemaName          string
+	TableName           string
+	IdentityColumnNames []string
+	Status              Status
+	Error               error
+}
+
+type Status string
+
+const (
+	StatusRequested  = Status("requested")
+	StatusInProgress = Status("in progress")
+	StatusCompleted  = Status("completed")
+)
+
+func (s *Snapshot) IsValid() bool {
+	return s != nil && s.SchemaName != "" && s.TableName != "" && len(s.IdentityColumnNames) > 0
+}
+
+func (s *Snapshot) MarkCompleted(err error) {
+	s.Status = StatusCompleted
+	s.Error = err
+}
+
+func (s *Snapshot) MarkInProgress() {
+	s.Status = StatusInProgress
+}
diff --git a/pkg/snapshot/store/mocks/mock_snapshot_store.go b/pkg/snapshot/store/mocks/mock_snapshot_store.go
new file mode 100644
index 0000000..4e90130
--- /dev/null
+++ b/pkg/snapshot/store/mocks/mock_snapshot_store.go
@@ -0,0 +1,27 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package mocks
+
+import (
+	"context"
+
+	"github.com/xataio/pgstream/pkg/snapshot"
+)
+
+type Store struct {
+	CreateSnapshotRequestFn func(context.Context, *snapshot.Snapshot) error
+	UpdateSnapshotRequestFn func(context.Context, *snapshot.Snapshot) error
+	GetSnapshotRequestsFn   func(ctx context.Context, status snapshot.Status) ([]*snapshot.Snapshot, error)
+}
+
+func (m *Store) CreateSnapshotRequest(ctx context.Context, s *snapshot.Snapshot) error {
+	return m.CreateSnapshotRequestFn(ctx, s)
+}
+
+func (m *Store) UpdateSnapshotRequest(ctx context.Context, s *snapshot.Snapshot) error {
+	return m.UpdateSnapshotRequestFn(ctx, s)
+}
+
+func (m *Store) GetSnapshotRequests(ctx context.Context, status snapshot.Status) ([]*snapshot.Snapshot, error) {
+	return m.GetSnapshotRequestsFn(ctx, status)
+}
diff --git a/pkg/snapshot/store/postgres/pg_snapshot_store.go b/pkg/snapshot/store/postgres/pg_snapshot_store.go
new file mode 100644
index 0000000..9adfe6c
--- /dev/null
+++ b/pkg/snapshot/store/postgres/pg_snapshot_store.go
@@ -0,0 +1,116 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package postgres
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/lib/pq"
+	"github.com/xataio/pgstream/internal/postgres"
+	"github.com/xataio/pgstream/pkg/snapshot"
+	"github.com/xataio/pgstream/pkg/snapshot/store"
+)
+
+type Store struct {
+	conn postgres.Querier
+}
+
+const queryLimit = 1000
+
+func NewStore(ctx context.Context, url string) (*Store, error) {
+	conn, err := postgres.NewConnPool(ctx, url)
+	if err != nil {
+		return nil, err
+	}
+
+	s := &Store{
+		conn: conn,
+	}
+
+	// create snapshots table if it doesn't exist
+	if err := s.createTable(ctx); err != nil {
+		return nil, fmt.Errorf("creating snapshots table: %w", err)
+	}
+
+	return s, nil
+}
+
+func (s *Store) Close() error {
+	return s.conn.Close(context.Background())
+}
+
+func (s *Store) CreateSnapshotRequest(ctx context.Context, req *snapshot.Snapshot) error {
+	query := fmt.Sprintf(`INSERT INTO %s (schema_name, table_name, identity_column_names, created_at, updated_at, status)
+	VALUES($1, $2, $3,'now()','now()','requested')`, snapshotsTable())
+	_, err := s.conn.Exec(ctx, query, req.SchemaName, req.TableName, pq.StringArray(req.IdentityColumnNames))
+	if err != nil {
+		return fmt.Errorf("error creating snapshot request: %w", err)
+	}
+	return nil
+}
+
+func (s *Store) UpdateSnapshotRequest(ctx context.Context, req *snapshot.Snapshot) error {
+	errStr := ""
+	if req.Error != nil {
+		errStr = req.Error.Error()
+	}
+	query := fmt.Sprintf(`UPDATE %s SET status = '%s', error = '%s', updated_at = 'now()'
+	WHERE schema_name = '%s' and table_name = '%s' and status != 'completed'`, snapshotsTable(), req.Status, errStr, req.SchemaName, req.TableName)
+	_, err := s.conn.Exec(ctx, query)
+	if err != nil {
+		return fmt.Errorf("error updating snapshot request: %w", err)
+	}
+	return nil
+}
+
+func (s *Store) GetSnapshotRequests(ctx context.Context, status snapshot.Status) ([]*snapshot.Snapshot, error) {
+	query := fmt.Sprintf(`SELECT schema_name,table_name,identity_column_names,status FROM %s
+	WHERE status = '%s' ORDER BY req_id ASC LIMIT %d`, snapshotsTable(), status, queryLimit)
+	rows, err := s.conn.Query(ctx, query)
+	if err != nil {
+		return nil, fmt.Errorf("error getting snapshot requests: %w", err)
+	}
+	defer rows.Close()
+
+	snapshots := []*snapshot.Snapshot{}
+	for rows.Next() {
+		snapshot := &snapshot.Snapshot{}
+		if err := rows.Scan(&snapshot.SchemaName, &snapshot.TableName, &snapshot.IdentityColumnNames, &snapshot.Status); err != nil {
+			return nil, fmt.Errorf("scanning snapshot row: %w", err)
+		}
+
+		snapshots = append(snapshots, snapshot)
+	}
+
+	return snapshots, nil
+}
+
+func (s *Store) createTable(ctx context.Context) error {
+	createQuery := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s(
+	req_id SERIAL PRIMARY KEY,
+	schema_name TEXT,
+	table_name TEXT,
+	identity_column_names TEXT[],
+	created_at TIMESTAMP WITH TIME ZONE,
+	updated_at TIMESTAMP WITH TIME ZONE,
+	status TEXT CHECK (status IN ('requested', 'in progress', 'completed')),
+	error TEXT )`, snapshotsTable())
+	_, err := s.conn.Exec(ctx, createQuery)
+	if err != nil {
+		return fmt.Errorf("error creating snapshots postgres table: %w", err)
+	}
+
+	uniqueIndexQuery := fmt.Sprintf(`CREATE UNIQUE INDEX IF NOT EXISTS schema_table_status_unique_index
+	ON %s(schema_name,table_name) WHERE status != 'completed'`, snapshotsTable())
+	_, err = s.conn.Exec(ctx, uniqueIndexQuery)
+	if err != nil {
+		return fmt.Errorf("error creating unique index on snapshots postgres table: %w", err)
+	}
+
+	return err
+}
+
+func snapshotsTable() string {
+	return fmt.Sprintf("%s.%s", pq.QuoteIdentifier(store.SchemaName), pq.QuoteIdentifier(store.TableName))
+}
diff --git a/pkg/snapshot/store/snapshot_store.go b/pkg/snapshot/store/snapshot_store.go
new file mode 100644
index 0000000..28e5432
--- /dev/null
+++ b/pkg/snapshot/store/snapshot_store.go
@@ -0,0 +1,20 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package store
+
+import (
+	"context"
+
+	"github.com/xataio/pgstream/pkg/snapshot"
+)
+
+type Store interface {
+	CreateSnapshotRequest(context.Context, *snapshot.Snapshot) error
+	UpdateSnapshotRequest(context.Context, *snapshot.Snapshot) error
+	GetSnapshotRequests(ctx context.Context, status snapshot.Status) ([]*snapshot.Snapshot, error)
+}
+
+const (
+	SchemaName = "pgstream"
+	TableName  = "snapshot_requests"
+)


Test patch:
diff --git a/pkg/schemalog/postgres/pg_schemalog_store_test.go b/pkg/schemalog/postgres/pg_schemalog_store_test.go
index fa1546e..02c2fb9 100644
--- a/pkg/schemalog/postgres/pg_schemalog_store_test.go
+++ b/pkg/schemalog/postgres/pg_schemalog_store_test.go
@@ -116,7 +116,7 @@ func TestStore_Ack(t *testing.T) {
 		{
 			name: "ok",
 			querier: &pgmocks.Querier{
-				ExecFn: func(_ context.Context, query string, args ...any) (pglib.CommandTag, error) {
+				ExecFn: func(_ context.Context, _ uint, query string, args ...any) (pglib.CommandTag, error) {
 					require.Len(t, args, 2)
 					require.Equal(t, args[0], testID.String())
 					require.Equal(t, args[1], testSchema)
@@ -133,7 +133,7 @@ func TestStore_Ack(t *testing.T) {
 		{
 			name: "error - executing update query",
 			querier: &pgmocks.Querier{
-				ExecFn: func(_ context.Context, query string, args ...any) (pglib.CommandTag, error) {
+				ExecFn: func(_ context.Context, _ uint, query string, args ...any) (pglib.CommandTag, error) {
 					return pglib.CommandTag{CommandTag: pgconn.NewCommandTag("")}, errTest
 				},
 			},
diff --git a/pkg/snapshot/store/postgres/pg_snapshot_store_test.go b/pkg/snapshot/store/postgres/pg_snapshot_store_test.go
new file mode 100644
index 0000000..e771b3d
--- /dev/null
+++ b/pkg/snapshot/store/postgres/pg_snapshot_store_test.go
@@ -0,0 +1,366 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package postgres
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/lib/pq"
+	"github.com/stretchr/testify/require"
+	"github.com/xataio/pgstream/internal/postgres"
+	postgresmocks "github.com/xataio/pgstream/internal/postgres/mocks"
+	"github.com/xataio/pgstream/pkg/snapshot"
+)
+
+func TestStore_CreateSnapshotRequest(t *testing.T) {
+	t.Parallel()
+
+	testSnapshot := &snapshot.Snapshot{
+		SchemaName:          "test-schema",
+		TableName:           "test-table",
+		IdentityColumnNames: []string{"id"},
+	}
+
+	errTest := errors.New("oh noes")
+
+	tests := []struct {
+		name    string
+		querier postgres.Querier
+
+		wantErr error
+	}{
+		{
+			name: "ok",
+			querier: &postgresmocks.Querier{
+				ExecFn: func(ctx context.Context, _ uint, s string, a ...any) (postgres.CommandTag, error) {
+					wantQuery := fmt.Sprintf(`INSERT INTO %s (schema_name, table_name, identity_column_names, created_at, updated_at, status)
+	VALUES($1, $2, $3,'now()','now()','requested')`, snapshotsTable())
+					require.Equal(t, wantQuery, s)
+					wantAttr := []any{testSnapshot.SchemaName, testSnapshot.TableName, pq.StringArray(testSnapshot.IdentityColumnNames)}
+					require.Equal(t, wantAttr, a)
+					return postgres.CommandTag{}, nil
+				},
+			},
+
+			wantErr: nil,
+		},
+		{
+			name: "error - creating snapshot",
+			querier: &postgresmocks.Querier{
+				ExecFn: func(ctx context.Context, _ uint, s string, a ...any) (postgres.CommandTag, error) {
+					return postgres.CommandTag{}, errTest
+				},
+			},
+
+			wantErr: errTest,
+		},
+	}
+
+	for _, tc := range tests {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			store := &Store{
+				conn: tc.querier,
+			}
+			err := store.CreateSnapshotRequest(context.Background(), testSnapshot)
+			require.ErrorIs(t, err, tc.wantErr)
+		})
+	}
+}
+
+func TestStore_UpdateSnapshotRequest(t *testing.T) {
+	t.Parallel()
+
+	testSnapshot := snapshot.Snapshot{
+		SchemaName:          "test-schema",
+		TableName:           "test-table",
+		IdentityColumnNames: []string{"id"},
+		Status:              snapshot.StatusInProgress,
+	}
+
+	errTest := errors.New("oh noes")
+
+	tests := []struct {
+		name    string
+		querier postgres.Querier
+		req     *snapshot.Snapshot
+
+		wantErr error
+	}{
+		{
+			name: "ok - update without error string",
+			querier: &postgresmocks.Querier{
+				ExecFn: func(ctx context.Context, _ uint, s string, a ...any) (postgres.CommandTag, error) {
+					wantQuery := fmt.Sprintf(`UPDATE %s SET status = '%s', error = '%s', updated_at = 'now()'
+	WHERE schema_name = '%s' and table_name = '%s' and status != 'completed'`,
+						snapshotsTable(), snapshot.StatusInProgress, "", testSnapshot.SchemaName, testSnapshot.TableName)
+					require.Equal(t, wantQuery, s)
+					require.Empty(t, a)
+					return postgres.CommandTag{}, nil
+				},
+			},
+			req: &testSnapshot,
+
+			wantErr: nil,
+		},
+		{
+			name: "ok - update with error string",
+			querier: &postgresmocks.Querier{
+				ExecFn: func(ctx context.Context, _ uint, s string, a ...any) (postgres.CommandTag, error) {
+					wantQuery := fmt.Sprintf(`UPDATE %s SET status = '%s', error = '%s', updated_at = 'now()'
+	WHERE schema_name = '%s' and table_name = '%s' and status != 'completed'`,
+						snapshotsTable(), snapshot.StatusInProgress, errTest.Error(), testSnapshot.SchemaName, testSnapshot.TableName)
+					require.Equal(t, wantQuery, s)
+					require.Empty(t, a)
+					return postgres.CommandTag{}, nil
+				},
+			},
+			req: func() *snapshot.Snapshot {
+				s := testSnapshot
+				s.Error = errTest
+				return &s
+			}(),
+
+			wantErr: nil,
+		},
+		{
+			name: "error - updating snapshot",
+			querier: &postgresmocks.Querier{
+				ExecFn: func(ctx context.Context, _ uint, s string, a ...any) (postgres.CommandTag, error) {
+					return postgres.CommandTag{}, errTest
+				},
+			},
+			req: &testSnapshot,
+
+			wantErr: errTest,
+		},
+	}
+
+	for _, tc := range tests {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			store := &Store{
+				conn: tc.querier,
+			}
+			err := store.UpdateSnapshotRequest(context.Background(), tc.req)
+			require.ErrorIs(t, err, tc.wantErr)
+		})
+	}
+}
+
+func TestStore_GetSnapshotRequests(t *testing.T) {
+	t.Parallel()
+
+	testSnapshot := &snapshot.Snapshot{
+		SchemaName:          "test-schema",
+		TableName:           "test-table",
+		IdentityColumnNames: []string{"id"},
+		Status:              snapshot.StatusInProgress,
+	}
+
+	errTest := errors.New("oh noes")
+
+	tests := []struct {
+		name    string
+		querier postgres.Querier
+
+		wantSnapshots []*snapshot.Snapshot
+		wantErr       error
+	}{
+		{
+			name: "ok - no results",
+			querier: &postgresmocks.Querier{
+				QueryFn: func(ctx context.Context, query string, args ...any) (postgres.Rows, error) {
+					wantQuery := fmt.Sprintf(`SELECT schema_name,table_name,identity_column_names,status FROM %s
+	WHERE status = '%s' ORDER BY req_id ASC LIMIT %d`, snapshotsTable(), snapshot.StatusInProgress, queryLimit)
+					require.Equal(t, wantQuery, query)
+					return &postgresmocks.Rows{
+						CloseFn: func() {},
+						NextFn:  func(_ uint) bool { return false },
+					}, nil
+				},
+			},
+
+			wantSnapshots: []*snapshot.Snapshot{},
+			wantErr:       nil,
+		},
+		{
+			name: "ok",
+			querier: &postgresmocks.Querier{
+				QueryFn: func(ctx context.Context, query string, args ...any) (postgres.Rows, error) {
+					wantQuery := fmt.Sprintf(`SELECT schema_name,table_name,identity_column_names,status FROM %s
+	WHERE status = '%s' ORDER BY req_id ASC LIMIT %d`, snapshotsTable(), snapshot.StatusInProgress, queryLimit)
+					require.Equal(t, wantQuery, query)
+					return &postgresmocks.Rows{
+						CloseFn: func() {},
+						NextFn:  func(i uint) bool { return i == 1 },
+						ScanFn: func(dest ...any) error {
+							require.Len(t, dest, 4)
+							schemaName, ok := dest[0].(*string)
+							require.True(t, ok)
+							*schemaName = testSnapshot.SchemaName
+
+							tableName, ok := dest[1].(*string)
+							require.True(t, ok)
+							*tableName = testSnapshot.TableName
+
+							idColumns, ok := dest[2].(*[]string)
+							require.True(t, ok)
+							*idColumns = testSnapshot.IdentityColumnNames
+
+							status, ok := dest[3].(*snapshot.Status)
+							require.True(t, ok)
+							*status = testSnapshot.Status
+							return nil
+						},
+					}, nil
+				},
+			},
+
+			wantSnapshots: []*snapshot.Snapshot{
+				testSnapshot,
+			},
+			wantErr: nil,
+		},
+		{
+			name: "error - querying",
+			querier: &postgresmocks.Querier{
+				QueryFn: func(ctx context.Context, query string, args ...any) (postgres.Rows, error) {
+					return nil, errTest
+				},
+			},
+
+			wantSnapshots: nil,
+			wantErr:       errTest,
+		},
+		{
+			name: "error - scanning row",
+			querier: &postgresmocks.Querier{
+				QueryFn: func(ctx context.Context, query string, args ...any) (postgres.Rows, error) {
+					wantQuery := fmt.Sprintf(`SELECT schema_name,table_name,identity_column_names,status FROM %s
+	WHERE status = '%s' ORDER BY req_id ASC LIMIT %d`, snapshotsTable(), snapshot.StatusInProgress, queryLimit)
+					require.Equal(t, wantQuery, query)
+					return &postgresmocks.Rows{
+						CloseFn: func() {},
+						NextFn:  func(i uint) bool { return i == 1 },
+						ScanFn: func(dest ...any) error {
+							return errTest
+						},
+					}, nil
+				},
+			},
+
+			wantSnapshots: nil,
+			wantErr:       errTest,
+		},
+	}
+
+	for _, tc := range tests {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			store := &Store{
+				conn: tc.querier,
+			}
+			snapshots, err := store.GetSnapshotRequests(context.Background(), snapshot.StatusInProgress)
+			require.ErrorIs(t, err, tc.wantErr)
+			require.Equal(t, tc.wantSnapshots, snapshots)
+		})
+	}
+}
+
+func TestStore_createTable(t *testing.T) {
+	t.Parallel()
+
+	errTest := errors.New("oh noes")
+
+	tests := []struct {
+		name    string
+		querier postgres.Querier
+
+		wantErr error
+	}{
+		{
+			name: "ok",
+			querier: &postgresmocks.Querier{
+				ExecFn: func(ctx context.Context, i uint, s string, a ...any) (postgres.CommandTag, error) {
+					switch i {
+					case 1:
+						wantQuery := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s(
+	req_id SERIAL PRIMARY KEY,
+	schema_name TEXT,
+	table_name TEXT,
+	identity_column_names TEXT[],
+	created_at TIMESTAMP WITH TIME ZONE,
+	updated_at TIMESTAMP WITH TIME ZONE,
+	status TEXT CHECK (status IN ('requested', 'in progress', 'completed')),
+	error TEXT )`, snapshotsTable())
+						require.Equal(t, wantQuery, s)
+						require.Empty(t, a)
+					case 2:
+						wantQuery := fmt.Sprintf(`CREATE UNIQUE INDEX IF NOT EXISTS schema_table_status_unique_index
+	ON %s(schema_name,table_name) WHERE status != 'completed'`, snapshotsTable())
+						require.Equal(t, wantQuery, s)
+						require.Empty(t, a)
+					default:
+						return postgres.CommandTag{}, fmt.Errorf("unexpected Exec call: %d", i)
+					}
+
+					return postgres.CommandTag{}, nil
+				},
+			},
+			wantErr: nil,
+		},
+		{
+			name: "error - creating table",
+			querier: &postgresmocks.Querier{
+				ExecFn: func(ctx context.Context, i uint, s string, a ...any) (postgres.CommandTag, error) {
+					switch i {
+					case 1:
+						return postgres.CommandTag{}, errTest
+					default:
+						return postgres.CommandTag{}, fmt.Errorf("unexpected Exec call: %d", i)
+					}
+				},
+			},
+			wantErr: errTest,
+		},
+		{
+			name: "error - creating index",
+			querier: &postgresmocks.Querier{
+				ExecFn: func(ctx context.Context, i uint, s string, a ...any) (postgres.CommandTag, error) {
+					switch i {
+					case 1:
+						return postgres.CommandTag{}, nil
+					case 2:
+						return postgres.CommandTag{}, errTest
+					default:
+						return postgres.CommandTag{}, fmt.Errorf("unexpected Exec call: %d", i)
+					}
+				},
+			},
+			wantErr: errTest,
+		},
+	}
+
+	for _, tc := range tests {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			store := Store{
+				conn: tc.querier,
+			}
+			err := store.createTable(context.Background())
+			require.ErrorIs(t, err, tc.wantErr)
+		})
+	}
+}