feat: support assistant attachment handling in API

- refactor database schema
- support assistant attachment
- support  assistant embedding
- start working on RAG

Makefile

@@ -95,6 +95,10 @@ migrate-down:
 	@echo "Migrating down..."
 	@migrate -path database/migrations -database "$(DATABASE_URL)" down
 
+migrate-drop:
+	@echo "Migrating drop..."
+	@migrate -path database/migrations -database "$(DATABASE_URL)" drop
+
 psql:
 	@echo "Connecting to database..."
 	@docker compose exec -it postgres psql -U ${DATABASE_USER} -d ${DATABASE_NAME}

database/migrations.go

@@ -2,15 +2,22 @@ package migrations
 
 import (
 	"context"
+	"fmt"
+	"strings"
 	"vibrain/internal/pkg/logger"
 
 	"github.com/golang-migrate/migrate/v4"
 	_ "github.com/golang-migrate/migrate/v4/database/postgres"
 	bindata "github.com/golang-migrate/migrate/v4/source/go_bindata"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/riverqueue/river/riverdriver/riverpgxv5"
+	"github.com/riverqueue/river/rivermigrate"
 )
 
 func Migrate(ctx context.Context, databaseURL string) {
-	logger.Default.Info("Migrating database")
+	logger.Default.Info("start migrating database")
+	migrateRiver(ctx, databaseURL)
+
 	s := bindata.Resource(AssetNames(), func(name string) ([]byte, error) {
 		return Asset(name)
 	})
@@ -25,11 +32,42 @@ func Migrate(ctx context.Context, databaseURL string) {
 	}
 	if err := m.Up(); err != nil {
 		if err.Error() == "no change" {
-			logger.Default.Info("No migration needed")
+			logger.Default.Info("No migration for normal db needed")
 			return
 		}
-		logger.Default.Error("Error while migrating", "err", err)
+		logger.Default.Fatal("Error while migrating", "err", err)
 		return
 	}
 	logger.Default.Info("Migration successful")
 }
+
+func migrateRiver(ctx context.Context, databaseURL string) {
+	logger.Default.Info("start migrating river")
+	dbPool, err := pgxpool.New(ctx, databaseURL)
+	if err != nil {
+		logger.Default.Fatal("migrateRiver failed to connect to database", "err", err)
+	}
+	defer dbPool.Close()
+
+	migrator := rivermigrate.New(riverpgxv5.New(dbPool), nil)
+	tx, err := dbPool.Begin(ctx)
+	if err != nil {
+		logger.Default.Fatal("migrateRiver failed to start transaction", "err", err)
+	}
+
+	defer func() {
+		if err := tx.Commit(ctx); err != nil {
+			logger.Default.Error("Failed to commit transaction", "err", err)
+		}
+	}()
+	res, err := migrator.MigrateTx(ctx, tx, rivermigrate.DirectionUp, &rivermigrate.MigrateOpts{
+		TargetVersion: -1,
+	})
+	if err != nil {
+		logger.Default.Fatal("migrateRiver failed to migrate", "err", err)
+	}
+	for _, version := range res.Versions {
+		logger.Default.Info(fmt.Sprintf("Migrated [%s] version %d", strings.ToUpper(string(res.Direction)), version.Version))
+	}
+	logger.Default.Info("migrate river successful")
+}

database/migrations/000002_create_users_database.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS users;

database/migrations/000002_create_users_database.up.sql

@@ -0,0 +1,15 @@
+CREATE TABLE IF NOT EXISTS users (
+    id SERIAL PRIMARY KEY,
+    uuid UUID UNIQUE NOT NULL DEFAULT gen_random_uuid(),
+    username VARCHAR(255),
+    password_hash TEXT,
+    email VARCHAR(255) UNIQUE,
+    github VARCHAR(255) UNIQUE,
+    google VARCHAR(255) UNIQUE,
+    telegram VARCHAR(255) UNIQUE,
+    activate_assistant_id UUID,
+    activate_thread_id UUID,
+    status VARCHAR(255) NOT NULL DEFAULT 'pending',
+    created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP
+);

database/migrations/000003_create_assistant_threads.down.sql

@@ -0,0 +1,6 @@
+DROP TABLE assistant_embedddings;
+DROP TABLE assistant_attachments;
+DROP TABLE assistant_messages;
+DROP TABLE assistant_threads;
+DROP TABLE assistants;
+DROP EXTENSION IF EXISTS vector;