refactor: remove pocketbase (#1138)

app/migration/files_migration.go

@@ -0,0 +1,56 @@
+package migration
+
+import (
+	"context"
+	"fmt"
+	"io/fs"
+	"os"
+	"path/filepath"
+
+	"github.com/SecurityBrewery/catalyst/app/database"
+	"github.com/SecurityBrewery/catalyst/app/database/sqlc"
+	"github.com/SecurityBrewery/catalyst/app/upload"
+)
+
+type filesMigration struct{}
+
+func newFilesMigration() func() (migration, error) {
+	return func() (migration, error) {
+		return filesMigration{}, nil
+	}
+}
+
+func (filesMigration) name() string { return "005_pocketbase_files_to_tusd" }
+
+func (filesMigration) up(ctx context.Context, queries *sqlc.Queries, dir string, uploader *upload.Uploader) error {
+	oldUploadDir := filepath.Join(dir, "storage")
+	if _, err := os.Stat(oldUploadDir); os.IsNotExist(err) {
+		// If the old upload directory does not exist, we assume no migration is needed.
+		return nil
+	}
+
+	oldUploadRoot, err := os.OpenRoot(oldUploadDir)
+	if err != nil {
+		return fmt.Errorf("open old uploads root: %w", err)
+	}
+
+	files, err := database.PaginateItems(ctx, func(ctx context.Context, offset, limit int64) ([]sqlc.ListFilesRow, error) {
+		return queries.ListFiles(ctx, sqlc.ListFilesParams{Limit: limit, Offset: offset})
+	})
+	if err != nil {
+		return fmt.Errorf("list files: %w", err)
+	}
+
+	for _, file := range files {
+		data, err := fs.ReadFile(oldUploadRoot.FS(), filepath.Join(file.ID, file.Blob))
+		if err != nil {
+			return fmt.Errorf("read file %s: %w", file.Blob, err)
+		}
+
+		if _, err := uploader.CreateFile(file.ID, file.Name, data); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

app/migration/migration.go

@@ -0,0 +1,49 @@
+package migration
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+
+	"github.com/SecurityBrewery/catalyst/app/database/sqlc"
+	"github.com/SecurityBrewery/catalyst/app/upload"
+)
+
+type migration interface {
+	name() string
+	up(ctx context.Context, queries *sqlc.Queries, dir string, uploader *upload.Uploader) error
+}
+
+func Apply(ctx context.Context, queries *sqlc.Queries, dir string, uploader *upload.Uploader) error {
+	currentVersion, err := version(ctx, queries.WriteDB)
+	if err != nil {
+		return err
+	}
+
+	slog.InfoContext(ctx, "Current database version", "version", currentVersion)
+
+	migrations, err := migrations(currentVersion)
+	if err != nil {
+		return fmt.Errorf("failed to get migrations: %w", err)
+	}
+
+	if len(migrations) == 0 {
+		slog.InfoContext(ctx, "No migrations to apply")
+
+		return nil
+	}
+
+	for _, m := range migrations {
+		slog.InfoContext(ctx, "Applying migration", "name", m.name())
+
+		if err := m.up(ctx, queries, dir, uploader); err != nil {
+			return fmt.Errorf("migration %s failed: %w", m.name(), err)
+		}
+	}
+
+	if err := setVersion(ctx, queries.WriteDB, currentVersion+len(migrations)); err != nil {
+		return err
+	}
+
+	return nil
+}

app/migration/migration_test.go

@@ -0,0 +1,22 @@
+package migration
+
+import (
+	"testing"
+
+	_ "github.com/mattn/go-sqlite3"
+	"github.com/stretchr/testify/require"
+
+	"github.com/SecurityBrewery/catalyst/app/database"
+	"github.com/SecurityBrewery/catalyst/app/upload"
+)
+
+func TestApply(t *testing.T) {
+	t.Parallel()
+
+	dir := t.TempDir()
+	queries := database.TestDB(t, dir)
+	uploader, err := upload.New(dir)
+	require.NoError(t, err)
+
+	require.NoError(t, Apply(t.Context(), queries, dir, uploader))
+}

app/migration/migrations.go

@@ -0,0 +1,34 @@
+package migration
+
+import "fmt"
+
+var migrationGenerators = []func() (migration, error){
+	newSQLMigration("000_create_pocketbase_tables"),
+	newSQLMigration("001_create_tables"),
+	newFilesMigration(),
+	newSQLMigration("002_create_defaultdata"),
+	newSQLMigration("003_create_groups"),
+}
+
+func migrations(version int) ([]migration, error) {
+	var migrations []migration
+
+	if version < 0 || version > len(migrationGenerators) {
+		return nil, fmt.Errorf("invalid migration version: %d", version)
+	}
+
+	if version == len(migrationGenerators) {
+		return migrations, nil // No migrations to apply
+	}
+
+	for _, migrationFunc := range migrationGenerators[version:] {
+		migration, err := migrationFunc()
+		if err != nil {
+			return nil, fmt.Errorf("failed to create migration: %w", err)
+		}
+
+		migrations = append(migrations, migration)
+	}
+
+	return migrations, nil
+}

app/migration/migrations_test.go

@@ -0,0 +1,32 @@
+package migration
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestMigrations_Success(t *testing.T) {
+	t.Parallel()
+
+	migs, err := migrations(0)
+	require.NoError(t, err)
+	require.Len(t, migs, len(migrationGenerators))
+}
+
+func TestMigrations_VersionOffset(t *testing.T) {
+	t.Parallel()
+
+	migs, err := migrations(1)
+	require.NoError(t, err)
+	require.Len(t, migs, len(migrationGenerators)-1)
+}
+
+func TestMigrations_Error(t *testing.T) {
+	t.Parallel()
+
+	migs, err := migrations(999) // Invalid version
+	require.Error(t, err)
+	require.Nil(t, migs)
+	require.Contains(t, err.Error(), "invalid migration version: 999")
+}

app/migration/sql.go

@@ -0,0 +1,42 @@
+package migration
+
+import (
+	"context"
+	"fmt"
+
+	sqlmigrations "github.com/SecurityBrewery/catalyst/app/database/migrations"
+	"github.com/SecurityBrewery/catalyst/app/database/sqlc"
+	"github.com/SecurityBrewery/catalyst/app/upload"
+)
+
+type sqlMigration struct {
+	sqlName string
+	upSQL   string
+}
+
+func newSQLMigration(name string) func() (migration, error) {
+	return func() (migration, error) {
+		up, err := sqlmigrations.Migrations.ReadFile(name + ".up.sql")
+		if err != nil {
+			return nil, fmt.Errorf("failed to read up migration file for %s: %w", name, err)
+		}
+
+		return &sqlMigration{
+			sqlName: name,
+			upSQL:   string(up),
+		}, nil
+	}
+}
+
+func (m sqlMigration) name() string {
+	return m.sqlName
+}
+
+func (m sqlMigration) up(ctx context.Context, queries *sqlc.Queries, _ string, _ *upload.Uploader) error {
+	_, err := queries.WriteDB.ExecContext(ctx, m.upSQL)
+	if err != nil {
+		return fmt.Errorf("migration %s up failed: %w", m.sqlName, err)
+	}
+
+	return nil
+}

app/migration/sql_test.go

@@ -0,0 +1,41 @@
+package migration
+
+import (
+	"testing"
+
+	_ "github.com/mattn/go-sqlite3"
+	"github.com/stretchr/testify/require"
+
+	"github.com/SecurityBrewery/catalyst/app/database"
+	"github.com/SecurityBrewery/catalyst/app/upload"
+)
+
+func TestSQLMigration_UpAndDown(t *testing.T) {
+	t.Parallel()
+
+	m := sqlMigration{
+		sqlName: "test_migration",
+		upSQL:   "CREATE TABLE test_table (id INTEGER PRIMARY KEY, name TEXT);",
+	}
+
+	dir := t.TempDir()
+	queries := database.TestDB(t, dir)
+	uploader, err := upload.New(dir)
+	require.NoError(t, err)
+
+	// Test up
+	require.NoError(t, m.up(t.Context(), queries, dir, uploader))
+
+	// Table should exist
+	_, err = queries.WriteDB.Exec("INSERT INTO test_table (name) VALUES ('foo')")
+	require.NoError(t, err)
+}
+
+func TestNewSQLMigration_FileNotFound(t *testing.T) {
+	t.Parallel()
+
+	f := newSQLMigration("does_not_exist")
+	_, err := f()
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "failed to read up migration file")
+}

app/migration/version.go

@@ -0,0 +1,27 @@
+package migration
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+)
+
+func version(ctx context.Context, db *sql.DB) (int, error) {
+	// get the current version of the database
+	var currentVersion int
+	if err := db.QueryRowContext(ctx, "PRAGMA user_version").Scan(&currentVersion); err != nil {
+		return 0, fmt.Errorf("failed to get current database version: %w", err)
+	}
+
+	return currentVersion, nil
+}
+
+func setVersion(ctx context.Context, db *sql.DB, version int) error {
+	// Update the database version after successful migration
+	_, err := db.ExecContext(ctx, fmt.Sprintf("PRAGMA user_version = %d", version))
+	if err != nil {
+		return fmt.Errorf("failed to update database version: %w", err)
+	}
+
+	return nil
+}

app/migration/version_test.go

@@ -0,0 +1,28 @@
+package migration
+
+import (
+	"database/sql"
+	"testing"
+
+	_ "github.com/mattn/go-sqlite3"
+	"github.com/stretchr/testify/require"
+)
+
+func TestVersionAndSetVersion(t *testing.T) {
+	t.Parallel()
+
+	db, err := sql.Open("sqlite3", ":memory:")
+	require.NoError(t, err, "failed to open in-memory db")
+	defer db.Close()
+
+	ver, err := version(t.Context(), db)
+	require.NoError(t, err, "failed to get version")
+	require.Equal(t, 0, ver, "expected version 0")
+
+	err = setVersion(t.Context(), db, 2)
+	require.NoError(t, err, "failed to set version")
+
+	ver, err = version(t.Context(), db)
+	require.NoError(t, err, "failed to get version after set")
+	require.Equal(t, 2, ver, "expected version 2")
+}