Merged in feature/unittests (pull request #9)

Feature/unittests

* addedmoretests

* easyquerytests

* fixotelinit

* preppedqueriestotestcontainer

* dbqueries

* splitintoqueryfiles

* covered the bases

cmd/queryRunner/main.go

@@ -3,7 +3,6 @@ package main
 import (
 	"context"
 	"log"
-	"os"
 	"queryorchestration/api/queue"
 	"queryorchestration/internal/database"
 	"queryorchestration/internal/database/repository"
@@ -20,10 +19,10 @@ import (
 func main() {
 	ctx := context.Background()
 
-	closeTracer := otel.Init(ctx)
+	closeTracer := otel.New(ctx)
 	defer closeTracer()
 
-	database.RunMigrations(os.Getenv("PWD"))
+	database.RunMigrations(&database.MigrationConfig{})
 
 	cfg, err := config.LoadDefaultConfig(ctx)
 	if err != nil {

cmd/queryService/main.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"log"
 	"net"
-	"os"
 	"queryorchestration/api/controllers"
 	serviceinterfaces "queryorchestration/api/serviceInterfaces"
 	"queryorchestration/internal/collector"
@@ -23,10 +22,10 @@ import (
 func main() {
 	ctx := context.Background()
 
-	closeTracer := otel.Init(ctx)
+	closeTracer := otel.New(ctx)
 	defer closeTracer()
 
-	database.RunMigrations(os.Getenv("PWD"))
+	database.RunMigrations(&database.MigrationConfig{})
 
 	host := "0.0.0.0"
 	port := 8080

database/migrations/20250108174833_create_fullactivequery_view.down.sql

@@ -0,0 +1 @@
+DROP VIEW fullActiveQueries;

database/migrations/20250108174833_create_fullactivequery_view.up.sql

@@ -0,0 +1,11 @@
+
+CREATE VIEW fullActiveQueries AS
+    SELECT DISTINCT q.id, q.type, q.activeVersion, q.latestVersion, c.config, ARRAY_AGG(DISTINCT r.requiredQueryId)::uuid[] as requiredIds
+    FROM queries AS q
+    LEFT JOIN queryConfigs AS c ON q.id = c.queryId
+    LEFT JOIN requiredQueries AS r ON q.id = r.queryId
+    WHERE (c.id is null or c.addedVersion >= q.activeVersion
+    and COALESCE(c.removedVersion, q.activeVersion - 1) < q.activeVersion)
+    and (r.id is null or r.addedVersion >= q.activeVersion
+    and COALESCE(r.removedVersion, q.activeVersion - 1) < q.activeVersion)
+    GROUP BY q.id, q.type, q.activeversion, q.latestversion, c.config;

database/queries/query.sql

@@ -6,31 +6,14 @@ INSERT INTO queryDeprecations (queryId) VALUES ($1);
 
 -- name: IsQueryDeprecated :one
 SELECT EXISTS (
-    SELECT 1 FROM queryDeprecations where queryId = $1 and removedAt is not null
+    SELECT 1 FROM queryDeprecations where queryId = $1 and removedAt is null LIMIT 1
 );
 
 -- name: GetQuery :one
-SELECT q.id, q.type, q.activeVersion, q.latestVersion, c.config, ARRAY_AGG(DISTINCT r.requiredQueryId) as requiredIds
-    FROM queries AS q
-    LEFT JOIN queryConfigs AS c ON q.id = c.queryId
-    LEFT JOIN requiredQueries AS r ON q.id = r.queryId
-    WHERE q.id = $1
-    and c.addedVersion >= q.activeVersion
-    and COALESCE(c.removedVersion, q.activeVersion - 1) < q.activeVersion
-    and r.addedVersion >= q.activeVersion
-    and COALESCE(r.removedVersion, q.activeVersion - 1) < q.activeVersion
-    GROUP BY q.id, c.config;
+SELECT id, type, activeVersion, latestVersion, config, requiredIds FROM fullActiveQueries WHERE id = $1;
 
 -- name: ListQueries :many
-SELECT q.id, q.type, q.activeVersion, q.latestVersion, c.config, ARRAY_AGG(r.requiredQueryId) AS requiredIds
-    FROM queries AS q
-    LEFT JOIN queryConfigs AS c ON q.id = c.queryId
-    LEFT JOIN requiredQueries AS r ON q.id = r.queryId
-    WHERE c.addedVersion >= q.activeVersion
-    and COALESCE(c.removedVersion, q.activeVersion - 1) < q.activeVersion
-    and r.addedVersion >= q.activeVersion
-    and COALESCE(r.removedVersion, q.activeVersion - 1) < q.activeVersion
-    GROUP BY q.id, c.config;
+SELECT id, type, activeVersion, latestVersion, config, requiredIds FROM fullActiveQueries;
 
 -- name: CreateQuery :one
 INSERT INTO queries (type) VALUES ($1) RETURNING id;

internal/database/config_test.go

@@ -0,0 +1,84 @@
+package database
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestMustGetConnectionString(t *testing.T) {
+	t.Setenv("DB_USER", "user")
+	t.Setenv("DB_PASS", "pass")
+	t.Setenv("DB_HOST", "host")
+	t.Setenv("DB_PORT", "port")
+	t.Setenv("DB_NAME", "name")
+
+	connStr := mustGetConnectionString()
+	assert.Equal(t, "postgres://user:pass@host:port/name?", connStr)
+}
+
+func TestMustGetConnectionStringNoSSL(t *testing.T) {
+	t.Setenv("DB_USER", "user")
+	t.Setenv("DB_PASS", "pass")
+	t.Setenv("DB_HOST", "host")
+	t.Setenv("DB_PORT", "port")
+	t.Setenv("DB_NAME", "name")
+	t.Setenv("DB_NOSSL", "1")
+
+	connStr := mustGetConnectionString()
+	assert.Equal(t, "postgres://user:pass@host:port/name?sslmode=disable", connStr)
+}
+
+func TestMustGetConnectionStringNoEnv(t *testing.T) {
+	t.Setenv("DB_USER", "")
+	t.Setenv("DB_PASS", "")
+	t.Setenv("DB_HOST", "")
+	t.Setenv("DB_PORT", "")
+	t.Setenv("DB_NAME", "")
+
+	assert.Panics(t, func() { mustGetConnectionString() })
+}
+
+func TestGetPoolConfig(t *testing.T) {
+	t.Setenv("DB_USER", "user")
+	t.Setenv("DB_PASS", "pass")
+	t.Setenv("DB_HOST", "host")
+	t.Setenv("DB_PORT", "5432")
+	t.Setenv("DB_NAME", "name")
+
+	config, err := getPoolConfig()
+	assert.Nil(t, err)
+	assert.Equal(t, config.ConnString(), mustGetConnectionString())
+}
+
+func TestGetPoolConfigInvalidPort(t *testing.T) {
+	t.Setenv("DB_USER", "user")
+	t.Setenv("DB_PASS", "pass")
+	t.Setenv("DB_HOST", "host")
+	t.Setenv("DB_PORT", "invalid_port")
+	t.Setenv("DB_NAME", "name")
+
+	_, err := getPoolConfig()
+	assert.EqualError(t, err, "cannot parse `postgres://user:xxxxxx@host:invalid_port/name?`: failed to parse as URL (invalid port \":invalid_port\" after host)")
+}
+
+func TestMustGetPoolConfig(t *testing.T) {
+	t.Setenv("DB_USER", "user")
+	t.Setenv("DB_PASS", "pass")
+	t.Setenv("DB_HOST", "host")
+	t.Setenv("DB_PORT", "5432")
+	t.Setenv("DB_NAME", "name")
+
+	config := mustGetPoolConfig()
+	assert.Equal(t, config.ConnString(), mustGetConnectionString())
+}
+
+func TestMustGetPoolConfigInvalidPort(t *testing.T) {
+	t.Setenv("DB_USER", "user")
+	t.Setenv("DB_PASS", "pass")
+	t.Setenv("DB_HOST", "host")
+	t.Setenv("DB_PORT", "invalid_port")
+	t.Setenv("DB_NAME", "name")
+
+	assert.Panics(t, func() { mustGetPoolConfig() })
+}

internal/database/connection.go

@@ -10,23 +10,70 @@ import (
 	"github.com/jackc/pgx/v5/pgxpool"
 )
 
-func createConnectionString() string {
-	driver := "postgres"
-	dbUser := env.GetPanic("DB_USER")
-	dbPass := env.GetPanic("DB_PASS")
-	dbHost := env.GetPanic("DB_HOST")
-	dbPort := env.GetPanic("DB_PORT")
-	dbName := env.GetPanic("DB_NAME")
-	opts := "sslmode=disable"
-	connStr := fmt.Sprintf("%s://%s:%s@%s:%s/%s?%s", driver, dbUser, dbPass, dbHost, dbPort, dbName, opts)
+type config struct {
+	user       string
+	password   string
+	host       string
+	port       string
+	name       string
+	disableSSL string
+}
 
-	return connStr
+func mustGetConfig() *config {
+	user := env.GetPanic("DB_USER")
+	pass := env.GetPanic("DB_PASS")
+	host := env.GetPanic("DB_HOST")
+	port := env.GetPanic("DB_PORT")
+	name := env.GetPanic("DB_NAME")
+
+	disableSSL, _ := env.Get("DB_NOSSL")
+
+	return &config{
+		user:       user,
+		password:   pass,
+		host:       host,
+		port:       port,
+		name:       name,
+		disableSSL: disableSSL,
+	}
+}
+
+func mustGetConnectionString() string {
+	driver := "postgres"
+	conf := mustGetConfig()
+	opts := ""
+
+	if conf.disableSSL == "1" {
+		opts += "sslmode=disable"
+	}
+
+	return fmt.Sprintf("%s://%s:%s@%s:%s/%s?%s", driver, conf.user, conf.password, conf.host, conf.port, conf.name, opts)
+}
+
+func getPoolConfig() (*pgxpool.Config, error) {
+	connStr := mustGetConnectionString()
+
+	config, err := pgxpool.ParseConfig(connStr)
+	if err != nil {
+		return nil, err
+	}
+
+	return config, nil
+}
+
+func mustGetPoolConfig() *pgxpool.Config {
+	config, err := getPoolConfig()
+	if err != nil {
+		log.Panicf("Unable to create database config: %v\n", err)
+	}
+
+	return config
 }
 
 func GetDBPool(ctx context.Context) *pgxpool.Pool {
-	connStr := createConnectionString()
+	config := mustGetPoolConfig()
 
-	pool, err := pgxpool.New(ctx, connStr)
+	pool, err := pgxpool.NewWithConfig(ctx, config)
 	if err != nil {
 		log.Panicf("Unable to create database pool: %v\n", err)
 	}
@@ -35,9 +82,9 @@ func GetDBPool(ctx context.Context) *pgxpool.Pool {
 }
 
 func GetDBConn(ctx context.Context) *pgx.Conn {
-	connStr := createConnectionString()
+	config := mustGetConnectionString()
 
-	conn, err := pgx.Connect(ctx, connStr)
+	conn, err := pgx.Connect(ctx, config)
 	if err != nil {
 		log.Panicf("Unable to connect to database: %v\n", err)
 	}

internal/database/connection_test.go

@@ -3,7 +3,6 @@ package database_test
 import (
 	"context"
 	"fmt"
-	"os"
 	"queryorchestration/internal/database"
 	"testing"
 
@@ -20,8 +19,16 @@ func TestDBConn(t *testing.T) {
 
 	conn := database.GetDBConn(ctx)
 	assert.NotNil(t, conn)
+}
+
+func TestDBConnNoDB(t *testing.T) {
+	ctx := context.Background()
+	t.Setenv("DB_USER", "user")
+	t.Setenv("DB_PASS", "pass")
+	t.Setenv("DB_HOST", "host")
+	t.Setenv("DB_PORT", "5432")
+	t.Setenv("DB_NAME", "name")
 
-	os.Setenv("DB_HOST", "")
 	assert.Panics(t, func() { database.GetDBConn(ctx) })
 }
 
@@ -32,9 +39,6 @@ func TestDBPool(t *testing.T) {
 
 	pool := database.GetDBPool(ctx)
 	assert.NotNil(t, pool)
-
-	os.Setenv("DB_HOST", "")
-	assert.Panics(t, func() { database.GetDBPool(ctx) })
 }
 
 type dbConfig struct {
@@ -92,11 +96,11 @@ func createDB(t *testing.T, ctx context.Context) (*dbConfig, testcontainers.Cont
 	config.Host = mappedHost
 	config.Port = mappedPort.Int()
 
-	os.Setenv("DB_USER", config.User)
-	os.Setenv("DB_PASS", config.Password)
-	os.Setenv("DB_HOST", config.Host)
-	os.Setenv("DB_PORT", fmt.Sprint(config.Port))
-	os.Setenv("DB_NAME", config.Name)
+	t.Setenv("DB_USER", config.User)
+	t.Setenv("DB_PASS", config.Password)
+	t.Setenv("DB_HOST", config.Host)
+	t.Setenv("DB_PORT", fmt.Sprint(config.Port))
+	t.Setenv("DB_NAME", config.Name)
 
 	return &config, container
 }

internal/database/migrations.go

@@ -4,8 +4,8 @@ import (
 	"database/sql"
 	"fmt"
 	"log"
+	"os"
 	"path"
-	"queryorchestration/internal/env"
 
 	"github.com/golang-migrate/migrate/v4"
 	"github.com/golang-migrate/migrate/v4/database/postgres"
@@ -13,13 +13,13 @@ import (
 )
 
 func createDB() {
-	dbUser := env.GetPanic("DB_USER")
-	dbPass := env.GetPanic("DB_PASS")
-	dbHost := env.GetPanic("DB_HOST")
-	dbPort := env.GetPanic("DB_PORT")
-	dbName := env.GetPanic("DB_NAME")
 	driver := "postgres"
-	connStr := fmt.Sprintf("%s://%s:%s@%s:%s?sslmode=disable", driver, dbUser, dbPass, dbHost, dbPort)
+	opts := ""
+	conf := mustGetConfig()
+	if conf.disableSSL == "1" {
+		opts += "sslmode=disable"
+	}
+	connStr := fmt.Sprintf("%s://%s:%s@%s:%s?%s", driver, conf.user, conf.password, conf.host, conf.port, opts)
 
 	db, err := sql.Open(driver, connStr)
 	if err != nil {
@@ -27,7 +27,7 @@ func createDB() {
 	}
 	defer db.Close()
 
-	result, err := db.Exec(fmt.Sprintf("SELECT 'CREATE DATABASE %s' WHERE NOT EXISTS (SELECT FROM pg_database WHERE datname = '%s')", dbName, dbName))
+	result, err := db.Exec(fmt.Sprintf("SELECT 'CREATE DATABASE %s' WHERE NOT EXISTS (SELECT FROM pg_database WHERE datname = '%s')", conf.name, conf.name))
 	if err != nil {
 		log.Panicf("Error creating database: %v", err)
 	}
@@ -36,16 +36,30 @@ func createDB() {
 	if err != nil {
 		log.Panic(err)
 	} else if rows == 0 {
-		log.Printf("Database already exists: %s", dbName)
+		log.Printf("Database already exists: %s", conf.name)
 	} else {
-		log.Printf("Database created: %s", dbName)
+		log.Printf("Database created: %s", conf.name)
 	}
 }
 
-func RunMigrations(basePath string) {
+type MigrationConfig struct {
+	BasePath         string
+	ConnectionString string
+}
+
+func RunMigrations(config *MigrationConfig) {
 	createDB()
 
-	connStr := createConnectionString()
+	connStr := config.ConnectionString
+	if connStr == "" {
+		connStr = mustGetConnectionString()
+	}
+
+	basePath := config.BasePath
+	if basePath == "" {
+		basePath = os.Getenv("PWD")
+	}
+
 	driver := "postgres"
 
 	db, err := sql.Open(driver, connStr)
@@ -55,12 +69,12 @@ func RunMigrations(basePath string) {
 	defer db.Close()
 
 	if err = db.Ping(); err != nil {
-		log.Panic("Failed to connect to database:", err)
+		log.Panic("failed to connect to database:", err)
 	}
 
 	conn, err := postgres.WithInstance(db, &postgres.Config{})
 	if err != nil {
-		log.Panic(err)
+		log.Panic("failed to create migrations: ", err)
 	}
 
 	migPath := "file://" + path.Join(basePath, "database/migrations")

internal/database/migrations_test.go

@@ -6,6 +6,8 @@ import (
 	"path"
 	"queryorchestration/internal/database"
 	"testing"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestRunMigrations(t *testing.T) {
@@ -13,5 +15,17 @@ func TestRunMigrations(t *testing.T) {
 	_, container := createDB(t, ctx)
 	defer container.Terminate(ctx)
 
-	database.RunMigrations(path.Join(os.Getenv("PWD"), "../.."))
+	t.Setenv("DB_NOSSL", "1")
+
+	database.RunMigrations(&database.MigrationConfig{
+		BasePath: path.Join(os.Getenv("PWD"), "../.."),
+	})
+}
+
+func TestRunMigrationsNoDB(t *testing.T) {
+	assert.Panics(t, func() {
+		database.RunMigrations(&database.MigrationConfig{
+			BasePath: path.Join(os.Getenv("PWD"), "../.."),
+		})
+	})
 }

internal/database/repository/collector_test.go

@@ -0,0 +1,30 @@
+package repository_test
+
+import (
+	"context"
+	"queryorchestration/internal/database"
+	"queryorchestration/internal/database/repository"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestCollector(t *testing.T) {
+	ctx := context.Background()
+	pool, container := createDB(t, ctx)
+	defer container.Terminate(ctx)
+	queries := repository.New(pool)
+
+	collectorID := database.MustToDBUUID(uuid.New())
+
+	collectorQueries, err := queries.GetCollectorQueries(ctx, collectorID)
+	assert.Nil(t, err)
+	assert.Equal(t, 0, len(collectorQueries))
+	assert.ElementsMatch(t, []repository.GetCollectorQueriesRow{}, collectorQueries)
+
+	jobID := database.MustToDBUUID(uuid.New())
+
+	_, err = queries.GetCollectorFromJobID(ctx, jobID)
+	assert.NotNil(t, err)
+}

internal/database/repository/db_test.go

@@ -0,0 +1,41 @@
+package repository_test
+
+import (
+	"context"
+	"queryorchestration/internal/database/repository"
+	"testing"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/pashagolub/pgxmock/v3"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestQueriesNew(t *testing.T) {
+	pool, err := pgxmock.NewPool()
+	if err != nil {
+		t.Fatalf("failed to open pgxmock database: %v", err)
+	}
+
+	queries := repository.New(pool)
+	assert.NotNil(t, queries)
+}
+
+func TestQueriesWithTx(t *testing.T) {
+	ctx := context.Background()
+
+	pool, err := pgxmock.NewPool()
+	if err != nil {
+		t.Fatalf("failed to open pgxmock database: %v", err)
+	}
+
+	queries := repository.New(pool)
+	assert.NotNil(t, queries)
+
+	pool.ExpectBeginTx(pgx.TxOptions{})
+
+	tx, err := pool.Begin(ctx)
+	assert.Nil(t, err)
+
+	txQueries := queries.WithTx(tx)
+	assert.NotNil(t, txQueries)
+}

internal/database/repository/models.go

@@ -92,6 +92,15 @@ type Collectorquerydependencytree struct {
 	Queryversion    int32
 }
 
+type Fullactivequery struct {
+	ID            pgtype.UUID
+	Type          Querytype
+	Activeversion int32
+	Latestversion int32
+	Config        []byte
+	Requiredids   []pgtype.UUID
+}
+
 type Query struct {
 	ID            pgtype.UUID
 	Latestversion int32

internal/database/repository/query.sql.go

@@ -62,30 +62,12 @@ func (q *Queries) DeprecateQuery(ctx context.Context, queryid pgtype.UUID) error
 }
 
 const getQuery = `-- name: GetQuery :one
-SELECT q.id, q.type, q.activeVersion, q.latestVersion, c.config, ARRAY_AGG(DISTINCT r.requiredQueryId) as requiredIds
-    FROM queries AS q
-    LEFT JOIN queryConfigs AS c ON q.id = c.queryId
-    LEFT JOIN requiredQueries AS r ON q.id = r.queryId
-    WHERE q.id = $1
-    and c.addedVersion >= q.activeVersion
-    and COALESCE(c.removedVersion, q.activeVersion - 1) < q.activeVersion
-    and r.addedVersion >= q.activeVersion
-    and COALESCE(r.removedVersion, q.activeVersion - 1) < q.activeVersion
-    GROUP BY q.id, c.config
+SELECT id, type, activeVersion, latestVersion, config, requiredIds FROM fullActiveQueries WHERE id = $1
 `
 
-type GetQueryRow struct {
-	ID            pgtype.UUID
-	Type          Querytype
-	Activeversion int32
-	Latestversion int32
-	Config        []byte
-	Requiredids   []pgtype.UUID
-}
-
-func (q *Queries) GetQuery(ctx context.Context, id pgtype.UUID) (GetQueryRow, error) {
+func (q *Queries) GetQuery(ctx context.Context, id pgtype.UUID) (Fullactivequery, error) {
 	row := q.db.QueryRow(ctx, getQuery, id)
-	var i GetQueryRow
+	var i Fullactivequery
 	err := row.Scan(
 		&i.ID,
 		&i.Type,
@@ -120,7 +102,7 @@ func (q *Queries) GetQueryConfig(ctx context.Context, arg GetQueryConfigParams)
 
 const isQueryDeprecated = `-- name: IsQueryDeprecated :one
 SELECT EXISTS (
-    SELECT 1 FROM queryDeprecations where queryId = $1 and removedAt is not null
+    SELECT 1 FROM queryDeprecations where queryId = $1 and removedAt is null LIMIT 1
 )
 `
 
@@ -132,35 +114,18 @@ func (q *Queries) IsQueryDeprecated(ctx context.Context, queryid pgtype.UUID) (b
 }
 
 const listQueries = `-- name: ListQueries :many
-SELECT q.id, q.type, q.activeVersion, q.latestVersion, c.config, ARRAY_AGG(r.requiredQueryId) AS requiredIds
-    FROM queries AS q
-    LEFT JOIN queryConfigs AS c ON q.id = c.queryId
-    LEFT JOIN requiredQueries AS r ON q.id = r.queryId
-    WHERE c.addedVersion >= q.activeVersion
-    and COALESCE(c.removedVersion, q.activeVersion - 1) < q.activeVersion
-    and r.addedVersion >= q.activeVersion
-    and COALESCE(r.removedVersion, q.activeVersion - 1) < q.activeVersion
-    GROUP BY q.id, c.config
+SELECT id, type, activeVersion, latestVersion, config, requiredIds FROM fullActiveQueries
 `
 
-type ListQueriesRow struct {
-	ID            pgtype.UUID
-	Type          Querytype
-	Activeversion int32
-	Latestversion int32
-	Config        []byte
-	Requiredids   []pgtype.UUID
-}
-
-func (q *Queries) ListQueries(ctx context.Context) ([]ListQueriesRow, error) {
+func (q *Queries) ListQueries(ctx context.Context) ([]Fullactivequery, error) {
 	rows, err := q.db.Query(ctx, listQueries)
 	if err != nil {
 		return nil, err
 	}
 	defer rows.Close()
-	var items []ListQueriesRow
+	var items []Fullactivequery
 	for rows.Next() {
-		var i ListQueriesRow
+		var i Fullactivequery
 		if err := rows.Scan(
 			&i.ID,
 			&i.Type,

internal/database/repository/query_test.go

@@ -0,0 +1,90 @@
+package repository_test
+
+import (
+	"context"
+	"log"
+	"queryorchestration/internal/database/repository"
+	"testing"
+
+	"github.com/jackc/pgx/v5/pgtype"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestQueries(t *testing.T) {
+	ctx := context.Background()
+	pool, container := createDB(t, ctx)
+	defer container.Terminate(ctx)
+	queries := repository.New(pool)
+
+	contextQueryID, err := queries.CreateQuery(ctx, repository.Querytype(repository.QuerytypeContextFull))
+	assert.Nil(t, err)
+	assert.True(t, contextQueryID.Valid)
+
+	jsonQueryID, err := queries.CreateQuery(ctx, repository.Querytype(repository.QuerytypeJsonExtractor))
+	assert.Nil(t, err)
+	assert.True(t, jsonQueryID.Valid)
+
+	jsonConfig := []byte("{\"path\": \"example_path\"}")
+
+	err = queries.CreateRequiredQuery(ctx, repository.CreateRequiredQueryParams{
+		Queryid:         jsonQueryID,
+		Requiredqueryid: contextQueryID,
+		Addedversion:    1,
+	})
+	assert.Nil(t, err)
+
+	err = queries.CreateQueryConfig(ctx, repository.CreateQueryConfigParams{
+		Queryid:      jsonQueryID,
+		Config:       jsonConfig,
+		Addedversion: 1,
+	})
+	assert.Nil(t, err)
+
+	jsonQuery, err := queries.GetQuery(ctx, jsonQueryID)
+	assert.Nil(t, err)
+	assert.EqualExportedValues(t, repository.Fullactivequery{
+		ID:            jsonQueryID,
+		Type:          repository.QuerytypeJsonExtractor,
+		Activeversion: 1,
+		Latestversion: 1,
+		Config:        jsonConfig,
+		Requiredids:   []pgtype.UUID{contextQueryID},
+	}, jsonQuery)
+
+	jsonQueryConfig, err := queries.GetQueryConfig(ctx, repository.GetQueryConfigParams{
+		Queryid:      jsonQueryID,
+		Addedversion: jsonQuery.Activeversion,
+	})
+	assert.Nil(t, err)
+	assert.Equal(t, jsonConfig, jsonQueryConfig.Config)
+
+	qs, err := queries.ListQueries(ctx)
+	assert.Nil(t, err)
+	assert.Equal(t, 2, len(qs))
+	log.Print(qs)
+	assert.ElementsMatch(t, []repository.Fullactivequery{
+		{ID: jsonQueryID,
+			Type:          repository.QuerytypeJsonExtractor,
+			Activeversion: 1,
+			Latestversion: 1,
+			Config:        jsonConfig,
+			Requiredids:   []pgtype.UUID{contextQueryID}},
+		{ID: contextQueryID,
+			Type:          repository.QuerytypeContextFull,
+			Activeversion: 1,
+			Latestversion: 1,
+			Config:        nil,
+			Requiredids:   []pgtype.UUID{{}}},
+	}, qs)
+
+	isDeprecated, err := queries.IsQueryDeprecated(ctx, jsonQueryID)
+	assert.Nil(t, err)
+	assert.False(t, isDeprecated)
+
+	err = queries.DeprecateQuery(ctx, jsonQueryID)
+	assert.Nil(t, err)
+
+	isDeprecated, err = queries.IsQueryDeprecated(ctx, jsonQueryID)
+	assert.Nil(t, err)
+	assert.True(t, isDeprecated)
+}

internal/database/repository/result_test.go

@@ -0,0 +1,68 @@
+package repository_test
+
+import (
+	"context"
+	"queryorchestration/internal/database"
+	"queryorchestration/internal/database/repository"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgtype"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestResults(t *testing.T) {
+	ctx := context.Background()
+	pool, container := createDB(t, ctx)
+	defer container.Terminate(ctx)
+	queries := repository.New(pool)
+
+	jsonQueryID, err := queries.CreateQuery(ctx, repository.Querytype(repository.QuerytypeJsonExtractor))
+	assert.Nil(t, err)
+
+	documentID := database.MustToDBUUID(uuid.New())
+
+	jsonQuery, err := queries.GetQuery(ctx, jsonQueryID)
+	assert.Nil(t, err)
+
+	cleanVersion := int32(1)
+	textVersion := int32(1)
+	jsonResultValue := "example_value"
+
+	jsonResultID, err := queries.SetResult(ctx, repository.SetResultParams{
+		Queryid:      jsonQueryID,
+		Documentid:   documentID,
+		Value:        jsonResultValue,
+		Cleanversion: cleanVersion,
+		Textversion:  textVersion,
+		Queryversion: jsonQuery.Activeversion,
+	})
+	assert.Nil(t, err)
+	assert.True(t, jsonResultID.Valid)
+
+	resultsByDoc, err := queries.ListResultsByDocumentID(ctx, repository.ListResultsByDocumentIDParams{
+		Documentid:   documentID,
+		Cleanversion: cleanVersion,
+		Textversion:  textVersion,
+	})
+	assert.Nil(t, err)
+	assert.Equal(t, 1, len(resultsByDoc))
+	assert.ElementsMatch(t, []repository.ListResultsByDocumentIDRow{
+		{
+			ID:           jsonResultID,
+			Queryid:      jsonQueryID,
+			Queryversion: jsonQuery.Activeversion,
+		},
+	}, resultsByDoc)
+
+	results, err := queries.ListResultValuesByID(ctx, []pgtype.UUID{jsonResultID})
+	assert.Nil(t, err)
+	assert.Equal(t, 1, len(results))
+	assert.ElementsMatch(t, []repository.ListResultValuesByIDRow{
+		{
+			ID:      jsonResultID,
+			Queryid: jsonQueryID,
+			Value:   jsonResultValue,
+		},
+	}, results)
+}

internal/database/repository/sqlquery_test.go

@@ -0,0 +1,69 @@
+package repository_test
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path"
+	"queryorchestration/internal/database"
+	"testing"
+
+	"github.com/docker/go-connections/nat"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/testcontainers/testcontainers-go"
+	"github.com/testcontainers/testcontainers-go/wait"
+)
+
+func createDB(t *testing.T, ctx context.Context) (*pgxpool.Pool, testcontainers.Container) {
+	port, err := nat.NewPort("tcp", "5432")
+	if err != nil {
+		t.Fatalf("Failed to create port: %v", err)
+	}
+
+	name := "queryorchestration"
+	pass := "pass"
+	user := "postgres"
+
+	req := testcontainers.ContainerRequest{
+		Image: "postgres:latest",
+		Env: map[string]string{
+			"POSTGRES_DB":       name,
+			"POSTGRES_USER":     user,
+			"POSTGRES_PASSWORD": pass,
+		},
+		ExposedPorts: []string{port.Port()},
+		WaitingFor:   wait.ForListeningPort(port),
+	}
+
+	container, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{
+		ContainerRequest: req,
+		Started:          true,
+	})
+	if err != nil {
+		t.Fatalf("Failed to start container: %v", err)
+	}
+
+	host, err := container.Host(ctx)
+	if err != nil {
+		t.Fatalf("Failed to extract host: %v", err)
+	}
+	mappedPort, err := container.MappedPort(ctx, port)
+	if err != nil {
+		t.Fatalf("Failed to extract port: %v", err)
+	}
+
+	t.Setenv("DB_USER", user)
+	t.Setenv("DB_PASS", pass)
+	t.Setenv("DB_HOST", host)
+	t.Setenv("DB_PORT", fmt.Sprint(mappedPort.Int()))
+	t.Setenv("DB_NAME", name)
+	t.Setenv("DB_NOSSL", "1")
+
+	database.RunMigrations(&database.MigrationConfig{
+		BasePath: path.Join(os.Getenv("PWD"), "../../.."),
+	})
+
+	pool := database.GetDBPool(ctx)
+
+	return pool, container
+}

internal/env/get_test.go

@@ -1,7 +1,6 @@
 package env_test
 
 import (
-	"os"
 	"queryorchestration/internal/env"
 	"testing"
 
@@ -12,7 +11,7 @@ func TestGet(t *testing.T) {
 	name := "test_get_env_var_does_not_exist"
 
 	value := "value"
-	os.Setenv(name, value)
+	t.Setenv(name, value)
 
 	returnValue, err := env.Get(name)
 
@@ -20,7 +19,7 @@ func TestGet(t *testing.T) {
 	assert.Nil(t, err)
 
 	value = ""
-	os.Setenv(name, value)
+	t.Setenv(name, value)
 
 	returnValue, err = env.Get(name)
 

internal/env/getfatal_test.go

@@ -1,7 +1,6 @@
 package env_test
 
 import (
-	"os"
 	"queryorchestration/internal/env"
 	"testing"
 
@@ -12,14 +11,14 @@ func TestGetPanic(t *testing.T) {
 	name := "test_get_env_var_does_not_exist"
 
 	value := "value"
-	os.Setenv(name, value)
+	t.Setenv(name, value)
 
 	returnValue := env.GetPanic(name)
 
 	assert.Equal(t, value, returnValue)
 
 	value = ""
-	os.Setenv(name, value)
+	t.Setenv(name, value)
 
 	assert.Panics(t, func() { env.GetPanic(name) })
 }

internal/export/service_test.go

@@ -0,0 +1,26 @@
+package export_test
+
+import (
+	"queryorchestration/internal/database"
+	"queryorchestration/internal/database/repository"
+	"queryorchestration/internal/export"
+	"testing"
+
+	"github.com/pashagolub/pgxmock/v3"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNew(t *testing.T) {
+	pool, err := pgxmock.NewPool()
+	if err != nil {
+		t.Fatalf("failed to open pgxmock database: %v", err)
+	}
+	queries := repository.New(pool)
+	db := &database.Connection{
+		Queries: queries,
+		Pool:    pool,
+	}
+
+	svc := export.New(db)
+	assert.NotNil(t, svc)
+}

internal/otel/service.go

@@ -10,7 +10,7 @@ import (
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
 )
 
-func Init(ctx context.Context) func() {
+func New(ctx context.Context) func() {
 	if os.Getenv("ENABLE_OTEL") != "1" {
 		log.Println("OpenTelemetry is disabled. Set ENABLE_OTEL to enable.")
 		return func() {} // No-op shutdown function

internal/otel/service_test.go

@@ -0,0 +1,16 @@
+package otel_test
+
+import (
+	"context"
+	"queryorchestration/internal/otel"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNew(t *testing.T) {
+	ctx := context.Background()
+
+	svc := otel.New(ctx)
+	assert.NotNil(t, svc)
+}

internal/query/create_test.go

@@ -30,10 +30,8 @@ func TestCreate(t *testing.T) {
 
 	config := "{\"path\":\"example_path\"}"
 	q := query.Query{
-		ID:            uuid.New(),
-		Type:          queryprocessor.TypeJsonExtractor,
-		ActiveVersion: int32(1),
-		LatestVersion: int32(1),
+		ID:   uuid.New(),
+		Type: queryprocessor.TypeJsonExtractor,
 		RequiredQueryIDs: []uuid.UUID{
 			uuid.New(),
 		},

internal/query/createprivate_test.go

@@ -0,0 +1,122 @@
+package query
+
+import (
+	"context"
+	"queryorchestration/internal/database"
+	"queryorchestration/internal/database/repository"
+	queryprocessor "queryorchestration/internal/queryProcessor"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+	"github.com/pashagolub/pgxmock/v3"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestGetCreator(t *testing.T) {
+	pool, err := pgxmock.NewPool()
+	if err != nil {
+		t.Fatalf("failed to open pgxmock database: %v", err)
+	}
+	queries := repository.New(pool)
+	db := &database.Connection{
+		Queries: queries,
+		Pool:    pool,
+	}
+	svc := New(db)
+
+	queryType := queryprocessor.Type(queryprocessor.TypeContextFull)
+	creator, err := svc.getCreator(queryType)
+	assert.Nil(t, err)
+	assert.NotNil(t, creator)
+
+	queryType = queryprocessor.Type(queryprocessor.TypeJsonExtractor)
+	creator, err = svc.getCreator(queryType)
+	assert.Nil(t, err)
+	assert.NotNil(t, creator)
+
+	queryType = queryprocessor.Type(-1)
+	_, err = svc.getCreator(queryType)
+	assert.NotNil(t, err)
+}
+
+func TestParseCreateQuery(t *testing.T) {
+	cQuery := &queryprocessor.Create{
+		Type: queryprocessor.TypeContextFull,
+		RequiredQueryIDs: []uuid.UUID{
+			uuid.New(),
+		},
+		Config: "{\"key\":\"value\"}",
+	}
+
+	resultQuery, err := parseCreateQuery(cQuery)
+	assert.Nil(t, err)
+	assert.EqualExportedValues(t, createQuery{
+		Type:             repository.QuerytypeContextFull,
+		RequiredQueryIDs: database.MustToDBUUIDArray(cQuery.RequiredQueryIDs),
+		Config:           []byte(cQuery.Config),
+	}, *resultQuery)
+}
+
+func TestParseCreateQueryInvalidType(t *testing.T) {
+	cQuery := &queryprocessor.Create{
+		Type: queryprocessor.Type(-1),
+		RequiredQueryIDs: []uuid.UUID{
+			uuid.New(),
+		},
+		Config: "{\"key\":\"value\"}",
+	}
+
+	_, err := parseCreateQuery(cQuery)
+	assert.EqualError(t, err, "invalid database query type")
+}
+
+func TestSubmitCreate(t *testing.T) {
+	ctx := context.Background()
+
+	pool, err := pgxmock.NewPool()
+	if err != nil {
+		t.Fatalf("failed to open pgxmock database: %v", err)
+	}
+	queries := repository.New(pool)
+	db := &database.Connection{
+		Queries: queries,
+		Pool:    pool,
+	}
+	svc := New(db)
+
+	config := "{\"path\":\"example_path\"}"
+	q := Query{
+		ID:   uuid.New(),
+		Type: queryprocessor.TypeJsonExtractor,
+		RequiredQueryIDs: []uuid.UUID{
+			uuid.New(),
+		},
+		Config: config,
+	}
+	create := &queryprocessor.Create{
+		Type:             q.Type,
+		RequiredQueryIDs: q.RequiredQueryIDs,
+		Config:           q.Config,
+	}
+
+	dbType, err := queryprocessor.ToDBQueryType(create.Type)
+	assert.Nil(t, err)
+
+	pool.ExpectBeginTx(pgx.TxOptions{})
+	pool.ExpectQuery("name: CreateQuery :one").WithArgs(dbType).WillReturnRows(
+		pgxmock.NewRows([]string{"id"}).
+			AddRow(database.MustToDBUUID(q.ID)),
+	)
+	for _, req := range create.RequiredQueryIDs {
+		pool.ExpectExec("name: CreateRequiredQuery :exec").WithArgs(database.MustToDBUUID(q.ID), database.MustToDBUUID(req), int32(1)).
+			WillReturnResult(pgxmock.NewResult("", 1))
+	}
+	pool.ExpectExec("name: CreateQueryConfig :exec").WithArgs(database.MustToDBUUID(q.ID), []byte(create.Config), int32(1)).
+		WillReturnResult(pgxmock.NewResult("", 1))
+	pool.ExpectCommit()
+
+	id, err := svc.submitCreate(ctx, create)
+	assert.Nil(t, err)
+	assert.Equal(t, q.ID, id)
+}

internal/query/get.go

@@ -24,10 +24,10 @@ func (s *Service) Get(ctx context.Context, id uuid.UUID) (*Query, error) {
 		return nil, err
 	}
 
-	return ParseDBGetQuery(&query)
+	return ParseFullActiveQuery(&query)
 }
 
-func ParseDBGetQuery(q *repository.GetQueryRow) (*Query, error) {
+func ParseFullActiveQuery(q *repository.Fullactivequery) (*Query, error) {
 	reqQueryIDs := database.MustToUUIDArray(q.Requiredids)
 
 	qType, err := queryprocessor.ParseDBType(q.Type)

internal/query/list.go

@@ -2,8 +2,6 @@ package query
 
 import (
 	"context"
-	"queryorchestration/internal/database"
-	"queryorchestration/internal/database/repository"
 	queryprocessor "queryorchestration/internal/queryProcessor"
 )
 
@@ -21,7 +19,7 @@ func (s *Service) List(ctx context.Context, filters ListFilters) ([]*Query, erro
 
 	queries := make([]*Query, len(dbQueries))
 	for index, query := range dbQueries {
-		q, err := ParseDBListQuery(&query)
+		q, err := ParseFullActiveQuery(&query)
 		if err != nil {
 			return nil, err
 		}
@@ -31,21 +29,3 @@ func (s *Service) List(ctx context.Context, filters ListFilters) ([]*Query, erro
 
 	return queries, nil
 }
-
-func ParseDBListQuery(q *repository.ListQueriesRow) (*Query, error) {
-	reqQueryIDs := database.MustToUUIDArray(q.Requiredids)
-
-	qType, err := queryprocessor.ParseDBType(q.Type)
-	if err != nil {
-		return nil, err
-	}
-
-	return &Query{
-		ID:               database.MustToUUID(q.ID),
-		ActiveVersion:    q.Activeversion,
-		LatestVersion:    q.Latestversion,
-		Type:             qType,
-		RequiredQueryIDs: reqQueryIDs,
-		Config:           string(q.Config),
-	}, nil
-}

internal/query/parse_test.go

@@ -34,8 +34,8 @@ func TestParseQuery(t *testing.T) {
 	}, *out)
 }
 
-func TestParseDBListQuery(t *testing.T) {
-	q := &repository.ListQueriesRow{
+func TestParseFullActiveQuery(t *testing.T) {
+	q := &repository.Fullactivequery{
 		ID:            database.MustToDBUUID(uuid.New()),
 		Type:          repository.QuerytypeContextFull,
 		Activeversion: int32(1),
@@ -46,33 +46,7 @@ func TestParseDBListQuery(t *testing.T) {
 		Config: []byte("example"),
 	}
 
-	out, err := query.ParseDBListQuery(q)
-	assert.Nil(t, err)
-	assert.EqualExportedValues(t, query.Query{
-		ID:            database.MustToUUID(q.ID),
-		Type:          queryprocessor.TypeContextFull,
-		ActiveVersion: q.Activeversion,
-		LatestVersion: q.Latestversion,
-		RequiredQueryIDs: []uuid.UUID{
-			database.MustToUUID(q.Requiredids[0]),
-		},
-		Config: string(q.Config),
-	}, *out)
-}
-
-func TestParseDBGetQuery(t *testing.T) {
-	q := &repository.GetQueryRow{
-		ID:            database.MustToDBUUID(uuid.New()),
-		Type:          repository.QuerytypeContextFull,
-		Activeversion: int32(1),
-		Latestversion: int32(2),
-		Requiredids: []pgtype.UUID{
-			database.MustToDBUUID(uuid.New()),
-		},
-		Config: []byte("example"),
-	}
-
-	out, err := query.ParseDBGetQuery(q)
+	out, err := query.ParseFullActiveQuery(q)
 	assert.Nil(t, err)
 	assert.EqualExportedValues(t, query.Query{
 		ID:            database.MustToUUID(q.ID),

internal/query/updateprivate_test.go

@@ -0,0 +1,72 @@
+package query
+
+import (
+	"context"
+	"queryorchestration/internal/database"
+	"queryorchestration/internal/database/repository"
+	queryprocessor "queryorchestration/internal/queryProcessor"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/pashagolub/pgxmock/v3"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestGetUpdator(t *testing.T) {
+	pool, err := pgxmock.NewPool()
+	if err != nil {
+		t.Fatalf("failed to open pgxmock database: %v", err)
+	}
+	queries := repository.New(pool)
+	db := &database.Connection{
+		Queries: queries,
+		Pool:    pool,
+	}
+	svc := New(db)
+
+	queryType := queryprocessor.Type(queryprocessor.TypeContextFull)
+	updator, err := svc.getUpdator(queryType)
+	assert.Nil(t, err)
+	assert.NotNil(t, updator)
+
+	queryType = queryprocessor.Type(queryprocessor.TypeJsonExtractor)
+	updator, err = svc.getUpdator(queryType)
+	assert.Nil(t, err)
+	assert.NotNil(t, updator)
+
+	queryType = queryprocessor.Type(-1)
+	_, err = svc.getUpdator(queryType)
+	assert.NotNil(t, err)
+}
+
+func TestSubmitUpdate(t *testing.T) {
+	ctx := context.Background()
+
+	pool, err := pgxmock.NewPool()
+	if err != nil {
+		t.Fatalf("failed to open pgxmock database: %v", err)
+	}
+	queries := repository.New(pool)
+	db := &database.Connection{
+		Queries: queries,
+		Pool:    pool,
+	}
+	svc := New(db)
+
+	config := "{\"path\":\"example_path\"}"
+	q := Query{
+		ID: uuid.New(),
+		RequiredQueryIDs: []uuid.UUID{
+			uuid.New(),
+		},
+		Config: config,
+	}
+	update := &queryprocessor.Update{
+		ID:               q.ID,
+		RequiredQueryIDs: q.RequiredQueryIDs,
+		Config:           q.Config,
+	}
+
+	err = svc.submitUpdate(ctx, update)
+	assert.Nil(t, err)
+}

internal/queryQueue/create.go

@@ -8,15 +8,7 @@ import (
 
 func (q *Queue) getUnsyncedQueries() {
 	for _, query := range q.collectorQueries {
-		isSynced := false
-		for _, result := range q.results {
-			if result.QueryID != query.ID || result.QueryVersion != query.Version {
-				continue
-			}
-			isSynced = true
-			break
-		}
-		if isSynced {
+		if q.isQuerySynced(query) {
 			continue
 		}
 
@@ -24,6 +16,16 @@ func (q *Queue) getUnsyncedQueries() {
 	}
 }
 
+func (q *Queue) isQuerySynced(query *queryprocessor.Query) bool {
+	for _, result := range q.results {
+		if result.QueryID == query.ID && result.QueryVersion == query.Version {
+			return true
+		}
+	}
+
+	return false
+}
+
 func (c *Queue) getCollectorQueries(ctx context.Context) error {
 	if c.collectorQueries != nil {
 		return nil

internal/queryQueue/create_test.go

@@ -0,0 +1,139 @@
+package queryqueue
+
+import (
+	"context"
+	"queryorchestration/internal/collector"
+	"queryorchestration/internal/database"
+	"queryorchestration/internal/database/repository"
+	queryprocessor "queryorchestration/internal/queryProcessor"
+	"queryorchestration/internal/result"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgtype"
+	"github.com/pashagolub/pgxmock/v3"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestGetUnsyncedQueries(t *testing.T) {
+	queryOne := &queryprocessor.Query{
+		ID:      uuid.New(),
+		Version: int32(1),
+	}
+	svc := Queue{
+		collectorQueries: []*queryprocessor.Query{
+			queryOne,
+		},
+		results: []*result.Result{
+			{ID: uuid.New(), QueryID: queryOne.ID, QueryVersion: queryOne.Version},
+		},
+	}
+
+	svc.getUnsyncedQueries()
+	assert.EqualExportedValues(t, []*queryprocessor.Query(nil), svc.unsyncedQueue)
+
+	svc.results = []*result.Result{}
+	svc.unsyncedQueue = []*queryprocessor.Query{}
+
+	svc.getUnsyncedQueries()
+	assert.EqualExportedValues(t, []*queryprocessor.Query{queryOne}, svc.unsyncedQueue)
+}
+
+func TestGetCollectorQueries(t *testing.T) {
+	ctx := context.Background()
+
+	pool, err := pgxmock.NewPool()
+	if err != nil {
+		t.Fatalf("failed to open pgxmock database: %v", err)
+	}
+	queries := repository.New(pool)
+	db := &database.Connection{
+		Queries: queries,
+		Pool:    pool,
+	}
+
+	svc := Queue{
+		db: db,
+		collector: &collector.Collector{
+			ID: uuid.New(),
+		},
+	}
+	dbCollectorID := database.MustToDBUUID(svc.collector.ID)
+
+	collectorQueries := []*queryprocessor.Query{
+		{ID: uuid.New(), Type: queryprocessor.TypeContextFull, RequiredQueryIDs: []uuid.UUID{}, Version: int32(1)},
+	}
+
+	rows := pgxmock.NewRows([]string{"collectorId", "queryId", "type", "queryVersion", "requiredIds"})
+	for _, q := range collectorQueries {
+		dbID := database.MustToDBUUID(q.ID)
+		dbReqIDs := database.MustToDBUUIDArray(q.RequiredQueryIDs)
+		ty, err := queryprocessor.ToDBNullQueryType(q.Type)
+		assert.Nil(t, err)
+		rows = rows.
+			AddRow(dbCollectorID, dbID, ty, pgtype.Int4{Int32: int32(q.Version), Valid: true}, dbReqIDs)
+	}
+
+	pool.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorID).WillReturnRows(rows)
+
+	svc.getCollectorQueries(ctx)
+	assert.EqualExportedValues(t, collectorQueries, svc.collectorQueries)
+}
+
+func TestIsQuerySynced(t *testing.T) {
+	query := &queryprocessor.Query{
+		ID:      uuid.New(),
+		Version: int32(1),
+	}
+	svc := Queue{
+		results: []*result.Result{
+			{QueryID: query.ID, QueryVersion: query.Version},
+		},
+	}
+
+	isSynced := svc.isQuerySynced(query)
+	assert.True(t, isSynced)
+}
+
+func TestIsQuerySyncedNoResult(t *testing.T) {
+	query := &queryprocessor.Query{
+		ID:      uuid.New(),
+		Version: int32(1),
+	}
+	svc := Queue{
+		results: []*result.Result{},
+	}
+
+	isSynced := svc.isQuerySynced(query)
+	assert.False(t, isSynced)
+}
+
+func TestIsQuerySyncedOldResult(t *testing.T) {
+	query := &queryprocessor.Query{
+		ID:      uuid.New(),
+		Version: int32(1),
+	}
+	svc := Queue{
+		results: []*result.Result{
+			{QueryID: query.ID, QueryVersion: query.Version - 1},
+		},
+	}
+
+	isSynced := svc.isQuerySynced(query)
+	assert.False(t, isSynced)
+}
+
+func TestIsQuerySyncedNewResult(t *testing.T) {
+	query := &queryprocessor.Query{
+		ID:      uuid.New(),
+		Version: int32(1),
+	}
+	svc := Queue{
+		results: []*result.Result{
+			{QueryID: query.ID, QueryVersion: query.Version + 1},
+		},
+	}
+
+	isSynced := svc.isQuerySynced(query)
+	assert.False(t, isSynced)
+}

internal/queryQueue/execute.go

@@ -2,11 +2,7 @@ package queryqueue
 
 import (
 	"context"
-	"fmt"
-	contextfull "queryorchestration/internal/contextFull"
 	"queryorchestration/internal/database"
-	"queryorchestration/internal/database/repository"
-	jsonextractor "queryorchestration/internal/jsonExtractor"
 	queryprocessor "queryorchestration/internal/queryProcessor"
 	"queryorchestration/internal/result"
 
@@ -71,21 +67,3 @@ func (q *Queue) executeQuery(ctx context.Context, qu *queryprocessor.Query) erro
 
 	return nil
 }
-
-func (q *Queue) getResultValue(res *repository.ListResultValuesByIDRow) (result.Value, error) {
-	var queryType queryprocessor.Type
-	for _, qu := range q.collectorQueries {
-		if qu.ID == database.MustToUUID(res.Queryid) {
-			queryType = qu.Type
-		}
-	}
-
-	switch queryType {
-	case queryprocessor.TypeJsonExtractor:
-		return jsonextractor.NewResult(res.Value), nil
-	case queryprocessor.TypeContextFull:
-		return contextfull.NewResult(res.Value), nil
-	default:
-		return nil, fmt.Errorf("attempting to process invalid query type")
-	}
-}

internal/queryQueue/execute_test.go

@@ -2,7 +2,6 @@ package queryqueue_test
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"queryorchestration/internal/collector"
 	"queryorchestration/internal/database"
@@ -18,7 +17,7 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-func TestQueue(t *testing.T) {
+func TestExecute(t *testing.T) {
 	ctx := context.Background()
 
 	pool, err := pgxmock.NewPool()
@@ -71,10 +70,7 @@ func TestQueue(t *testing.T) {
 	rows := pgxmock.NewRows([]string{"collectorId", "queryId", "type", "queryVersion", "requiredIds"})
 	for _, q := range collectorQueries {
 		dbID := database.MustToDBUUID(q.ID)
-		dbReqIDs := make([]pgtype.UUID, len(q.RequiredQueryIDs))
-		for index, id := range q.RequiredQueryIDs {
-			dbReqIDs[index] = database.MustToDBUUID(id)
-		}
+		dbReqIDs := database.MustToDBUUIDArray(q.RequiredQueryIDs)
 		ty, err := queryprocessor.ToDBNullQueryType(q.Type)
 		assert.Nil(t, err)
 		rows = rows.
@@ -181,43 +177,3 @@ func TestQueue(t *testing.T) {
 	err = q.Execute(ctx)
 	assert.Nil(t, err)
 }
-
-func TestQueueFail(t *testing.T) {
-	ctx := context.Background()
-
-	pool, err := pgxmock.NewPool()
-	if err != nil {
-		t.Fatalf("failed to open pgxmock database: %v", err)
-	}
-	queries := repository.New(pool)
-	db := &database.Connection{
-		Queries: queries,
-		Pool:    pool,
-	}
-	jobID := uuid.New()
-	dbJobID := database.MustToDBUUID(jobID)
-	collectorID := uuid.New()
-	dbCollectorID := database.MustToDBUUID(collectorID)
-
-	pool.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID).
-		WillReturnRows(
-			pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion"}).
-				AddRow(dbCollectorID, dbJobID, int32(1), int32(1)),
-		)
-
-	coll, err := collector.NewByJobId(ctx, db, jobID)
-	assert.Nil(t, err)
-
-	errr := "database failure"
-	pool.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorID).
-		WillReturnError(errors.New(errr))
-
-	results := []*result.Result{}
-
-	docID := uuid.New()
-	cleanVersion := int32(1)
-	textVersion := int32(1)
-
-	_, err = queryqueue.New(ctx, db, coll, results, docID, cleanVersion, textVersion)
-	assert.EqualError(t, err, errr)
-}

internal/queryQueue/executeprivate_test.go

@@ -0,0 +1,94 @@
+package queryqueue
+
+import (
+	"context"
+	"queryorchestration/internal/database"
+	"queryorchestration/internal/database/repository"
+	queryprocessor "queryorchestration/internal/queryProcessor"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgtype"
+	"github.com/pashagolub/pgxmock/v3"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestExecute(t *testing.T) {
+	ctx := context.Background()
+
+	pool, err := pgxmock.NewPool()
+	if err != nil {
+		t.Fatalf("failed to open pgxmock database: %v", err)
+	}
+	queries := repository.New(pool)
+	db := &database.Connection{
+		Queries: queries,
+		Pool:    pool,
+	}
+
+	expectedQueries := []*queryprocessor.Query{
+		{ID: uuid.New(), Type: queryprocessor.TypeContextFull, RequiredQueryIDs: []uuid.UUID{}, Version: int32(1)},
+		{ID: uuid.New(), Type: queryprocessor.TypeContextFull, RequiredQueryIDs: []uuid.UUID{}, Version: int32(2)},
+	}
+
+	q := &Queue{
+		unsyncedQueue: expectedQueries,
+		db:            db,
+		documentId:    uuid.New(),
+		cleanVersion:  int32(1),
+		textVersion:   int32(2),
+	}
+
+	pool.ExpectQuery("name: ListResultValuesByID :many").WithArgs([]pgtype.UUID{}).WillReturnRows(
+		pgxmock.NewRows([]string{"id", "queryId", "value"}),
+	)
+	pool.ExpectQuery("name: SetResult :one").WithArgs(database.MustToDBUUID(expectedQueries[0].ID), database.MustToDBUUID(q.documentId), pgxmock.AnyArg(), q.cleanVersion, q.textVersion, expectedQueries[0].Version).
+		WillReturnRows(
+			pgxmock.NewRows([]string{"id"}).AddRow(pgtype.UUID{}),
+		)
+
+	pool.ExpectQuery("name: ListResultValuesByID :many").WithArgs([]pgtype.UUID{}).WillReturnRows(
+		pgxmock.NewRows([]string{"id", "queryId", "value"}),
+	)
+	pool.ExpectQuery("name: SetResult :one").WithArgs(database.MustToDBUUID(expectedQueries[1].ID), database.MustToDBUUID(q.documentId), pgxmock.AnyArg(), q.cleanVersion, q.textVersion, expectedQueries[1].Version).
+		WillReturnRows(
+			pgxmock.NewRows([]string{"id"}).AddRow(pgtype.UUID{}),
+		)
+
+	err = q.Execute(ctx)
+	assert.Nil(t, err)
+}
+
+func TestExecuteQuery(t *testing.T) {
+	ctx := context.Background()
+
+	pool, err := pgxmock.NewPool()
+	if err != nil {
+		t.Fatalf("failed to open pgxmock database: %v", err)
+	}
+	queries := repository.New(pool)
+	db := &database.Connection{
+		Queries: queries,
+		Pool:    pool,
+	}
+
+	qu := &queryprocessor.Query{ID: uuid.New(), Type: queryprocessor.TypeContextFull, RequiredQueryIDs: []uuid.UUID{}, Version: int32(1)}
+
+	q := &Queue{
+		db:           db,
+		documentId:   uuid.New(),
+		cleanVersion: int32(1),
+		textVersion:  int32(2),
+	}
+
+	pool.ExpectQuery("name: ListResultValuesByID :many").WithArgs([]pgtype.UUID{}).WillReturnRows(
+		pgxmock.NewRows([]string{"id", "queryId", "value"}),
+	)
+	pool.ExpectQuery("name: SetResult :one").WithArgs(database.MustToDBUUID(qu.ID), database.MustToDBUUID(q.documentId), pgxmock.AnyArg(), q.cleanVersion, q.textVersion, qu.Version).
+		WillReturnRows(
+			pgxmock.NewRows([]string{"id"}).AddRow(pgtype.UUID{}),
+		)
+
+	err = q.executeQuery(ctx, qu)
+	assert.Nil(t, err)
+}

internal/queryQueue/result.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"fmt"
 	contextfull "queryorchestration/internal/contextFull"
+	"queryorchestration/internal/database"
+	"queryorchestration/internal/database/repository"
 	jsonextractor "queryorchestration/internal/jsonExtractor"
 	queryprocessor "queryorchestration/internal/queryProcessor"
 	"queryorchestration/internal/result"
@@ -51,3 +53,21 @@ func (q *Queue) getProcessor(queryType queryprocessor.Type) (queryprocessor.Proc
 		return nil, fmt.Errorf("attempting to process invalid query type")
 	}
 }
+
+func (q *Queue) getResultValue(res *repository.ListResultValuesByIDRow) (result.Value, error) {
+	var queryType queryprocessor.Type
+	for _, qu := range q.collectorQueries {
+		if qu.ID == database.MustToUUID(res.Queryid) {
+			queryType = qu.Type
+		}
+	}
+
+	switch queryType {
+	case queryprocessor.TypeJsonExtractor:
+		return jsonextractor.NewResult(res.Value), nil
+	case queryprocessor.TypeContextFull:
+		return contextfull.NewResult(res.Value), nil
+	default:
+		return nil, fmt.Errorf("attempting to process invalid query type")
+	}
+}

internal/queryQueue/resultprivate_test.go

@@ -0,0 +1,74 @@
+package queryqueue
+
+import (
+	"context"
+	"queryorchestration/internal/database"
+	"queryorchestration/internal/database/repository"
+	queryprocessor "queryorchestration/internal/queryProcessor"
+	"queryorchestration/internal/result"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgtype"
+	"github.com/pashagolub/pgxmock/v3"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSetResult(t *testing.T) {
+	ctx := context.Background()
+
+	pool, err := pgxmock.NewPool()
+	if err != nil {
+		t.Fatalf("failed to open pgxmock database: %v", err)
+	}
+	queries := repository.New(pool)
+	db := &database.Connection{
+		Queries: queries,
+		Pool:    pool,
+	}
+
+	q := &Queue{
+		db:           db,
+		documentId:   uuid.New(),
+		cleanVersion: int32(1),
+		textVersion:  int32(2),
+	}
+
+	qu := &queryprocessor.Query{ID: uuid.New(), Type: queryprocessor.TypeContextFull, RequiredQueryIDs: []uuid.UUID{}, Version: int32(1)}
+	resultValues := []result.Value{}
+
+	pool.ExpectQuery("name: SetResult :one").WithArgs(database.MustToDBUUID(qu.ID), database.MustToDBUUID(q.documentId), pgxmock.AnyArg(), q.cleanVersion, q.textVersion, qu.Version).
+		WillReturnRows(
+			pgxmock.NewRows([]string{"id"}).AddRow(pgtype.UUID{}),
+		)
+
+	err = q.setResult(ctx, qu, resultValues)
+	assert.Nil(t, err)
+}
+
+func TestGetProcessor(t *testing.T) {
+	q := Queue{}
+
+	qType := queryprocessor.Type(queryprocessor.TypeContextFull)
+
+	processor, err := q.getProcessor(qType)
+	assert.Nil(t, err)
+	assert.NotNil(t, processor)
+}
+
+func TestGetResultValue(t *testing.T) {
+	result := &repository.ListResultValuesByIDRow{
+		Queryid: database.MustToDBUUID(uuid.New()),
+		Value:   "EXAMPLE_VALUE",
+	}
+
+	q := &Queue{
+		collectorQueries: []*queryprocessor.Query{
+			{ID: database.MustToUUID(result.Queryid), Type: queryprocessor.TypeContextFull},
+		},
+	}
+
+	value, err := q.getResultValue(result)
+	assert.Nil(t, err)
+	assert.NotNil(t, value)
+}

internal/queryQueue/service_test.go

@@ -0,0 +1,145 @@
+package queryqueue_test
+
+import (
+	"context"
+	"errors"
+	"queryorchestration/internal/collector"
+	"queryorchestration/internal/database"
+	"queryorchestration/internal/database/repository"
+	queryprocessor "queryorchestration/internal/queryProcessor"
+	queryqueue "queryorchestration/internal/queryQueue"
+	"queryorchestration/internal/result"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgtype"
+	"github.com/pashagolub/pgxmock/v3"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestService(t *testing.T) {
+	ctx := context.Background()
+
+	pool, err := pgxmock.NewPool()
+	if err != nil {
+		t.Fatalf("failed to open pgxmock database: %v", err)
+	}
+	queries := repository.New(pool)
+	db := &database.Connection{
+		Queries: queries,
+		Pool:    pool,
+	}
+	jobID := uuid.New()
+	dbJobID := database.MustToDBUUID(jobID)
+	collectorID := uuid.New()
+	dbCollectorID := database.MustToDBUUID(collectorID)
+
+	pool.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID).
+		WillReturnRows(
+			pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion"}).
+				AddRow(dbCollectorID, dbJobID, int32(1), int32(1)),
+		)
+
+	coll, err := collector.NewByJobId(ctx, db, jobID)
+	assert.Nil(t, err)
+
+	queryOneID := uuid.New()
+	queryOneVersion := int32(1)
+	queryTwoID := uuid.New()
+	queryTwoVersion := int32(2)
+	queryThreeID := uuid.New()
+	queryThreeVersion := int32(3)
+	queryFourID := uuid.New()
+	queryFourVersion := int32(4)
+	queryFiveID := uuid.New()
+	queryFiveVersion := int32(5)
+	querySixID := uuid.New()
+	querySixVersion := int32(6)
+	contextID := uuid.New()
+	contextVersion := int32(1)
+	collectorQueries := []queryprocessor.Query{
+		{ID: contextID, Type: queryprocessor.TypeContextFull, RequiredQueryIDs: []uuid.UUID{}, Version: contextVersion},
+		{ID: queryOneID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{contextID}, Version: queryOneVersion},
+		{ID: queryTwoID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{queryOneID}, Version: queryTwoVersion},
+		{ID: queryThreeID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{queryOneID}, Version: queryThreeVersion},
+		{ID: queryFourID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{contextID}, Version: queryFourVersion},
+		{ID: queryFiveID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{querySixID}, Version: queryFiveVersion},
+		{ID: querySixID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{contextID}, Version: querySixVersion},
+	}
+
+	rows := pgxmock.NewRows([]string{"collectorId", "queryId", "type", "queryVersion", "requiredIds"})
+	for _, q := range collectorQueries {
+		dbID := database.MustToDBUUID(q.ID)
+		dbReqIDs := database.MustToDBUUIDArray(q.RequiredQueryIDs)
+		ty, err := queryprocessor.ToDBNullQueryType(q.Type)
+		assert.Nil(t, err)
+		rows = rows.
+			AddRow(dbCollectorID, dbID, ty, pgtype.Int4{Int32: int32(q.Version), Valid: true}, dbReqIDs)
+	}
+
+	pool.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorID).WillReturnRows(rows)
+
+	contextResultID := uuid.New()
+	results := []*result.Result{
+		{ID: contextResultID, QueryID: contextID, QueryVersion: contextVersion},
+		{ID: uuid.New(), QueryID: queryFourID, QueryVersion: queryFourVersion},
+		{ID: uuid.New(), QueryID: querySixID, QueryVersion: querySixVersion - 1},
+		{ID: uuid.New(), QueryID: queryOneID, QueryVersion: queryOneVersion - 1},
+	}
+
+	expectedQueries := []*queryprocessor.Query{
+		{ID: querySixID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{contextID}, Version: querySixVersion},
+		{ID: queryFiveID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{querySixID}, Version: queryFiveVersion},
+		{ID: queryOneID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{contextID}, Version: queryOneVersion},
+		{ID: queryThreeID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{queryOneID}, Version: queryThreeVersion},
+		{ID: queryTwoID, Type: queryprocessor.TypeJsonExtractor, RequiredQueryIDs: []uuid.UUID{queryOneID}, Version: queryTwoVersion},
+	}
+
+	docID := uuid.New()
+	cleanVersion := int32(1)
+	textVersion := int32(1)
+
+	q, err := queryqueue.New(ctx, db, coll, results, docID, cleanVersion, textVersion)
+	assert.Nil(t, err)
+	assert.Equal(t, expectedQueries, q.GetQueue())
+}
+
+func TestQueueFail(t *testing.T) {
+	ctx := context.Background()
+
+	pool, err := pgxmock.NewPool()
+	if err != nil {
+		t.Fatalf("failed to open pgxmock database: %v", err)
+	}
+	queries := repository.New(pool)
+	db := &database.Connection{
+		Queries: queries,
+		Pool:    pool,
+	}
+	jobID := uuid.New()
+	dbJobID := database.MustToDBUUID(jobID)
+	collectorID := uuid.New()
+	dbCollectorID := database.MustToDBUUID(collectorID)
+
+	pool.ExpectQuery("name: GetCollectorFromJobID :one").WithArgs(dbJobID).
+		WillReturnRows(
+			pgxmock.NewRows([]string{"id", "jobId", "minCleanVersion", "minTextVersion"}).
+				AddRow(dbCollectorID, dbJobID, int32(1), int32(1)),
+		)
+
+	coll, err := collector.NewByJobId(ctx, db, jobID)
+	assert.Nil(t, err)
+
+	errr := "database failure"
+	pool.ExpectQuery("name: GetCollectorQueries :many").WithArgs(dbCollectorID).
+		WillReturnError(errors.New(errr))
+
+	results := []*result.Result{}
+
+	docID := uuid.New()
+	cleanVersion := int32(1)
+	textVersion := int32(1)
+
+	_, err = queryqueue.New(ctx, db, coll, results, docID, cleanVersion, textVersion)
+	assert.EqualError(t, err, errr)
+}

internal/queue/queue_test.go

@@ -0,0 +1,92 @@
+package queue_test
+
+import (
+	"context"
+	"fmt"
+	"queryorchestration/internal/queue"
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/service/sqs"
+	"github.com/docker/go-connections/nat"
+	"github.com/testcontainers/testcontainers-go"
+	"github.com/testcontainers/testcontainers-go/wait"
+)
+
+type queueConfig struct {
+	Container *testcontainers.Container
+	Config    *queue.QueueConfig
+}
+
+func createQueue(t *testing.T, ctx context.Context) (*queueConfig, func()) {
+	queueName := "test-queue"
+	region := "us-east-1"
+	alias := "localstack"
+	provider := credentials.NewStaticCredentialsProvider("test", "test", "")
+
+	port, err := nat.NewPort("tcp", "4566")
+	if err != nil {
+		t.Fatalf("Failed to create port: %v", err)
+	}
+
+	req := testcontainers.ContainerRequest{
+		Image: "localstack/localstack:latest",
+		Env: map[string]string{
+			"AWS_ACCESS_KEY_ID":      provider.Value.AccessKeyID,
+			"AWS_SECRET_ACCESS_KEY":  provider.Value.SecretAccessKey,
+			"AWS_SESSION_TOKEN":      provider.Value.SessionToken,
+			"AWS_DEFAULT_REGION":     region,
+			"AWS_REGION":             region,
+			"SERVICES":               "sqs",
+			"SKIP_SSL_CERT_DOWNLOAD": "1",
+			"LOCALSTACK_HOST":        alias,
+			"SQS_ENDPOINT_STRATEGY":  "path",
+		},
+		ExposedPorts: []string{port.Port()},
+		WaitingFor:   wait.ForListeningPort(port),
+	}
+
+	container, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{
+		ContainerRequest: req,
+		Started:          true,
+	})
+	if err != nil {
+		t.Fatalf("Failed to start container: %v", err)
+	}
+
+	host, err := container.Host(ctx)
+	if err != nil {
+		t.Fatalf("Failed to extract host: %v", err)
+	}
+	mappedPort, err := container.MappedPort(ctx, port)
+	if err != nil {
+		t.Fatalf("Failed to extract port: %v", err)
+	}
+
+	endpoint := fmt.Sprintf("http://%s:%s", host, mappedPort.Port())
+	cfg, err := config.LoadDefaultConfig(ctx, config.WithRegion(region), config.WithCredentialsProvider(provider), config.WithBaseEndpoint(endpoint))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	client := sqs.NewFromConfig(cfg)
+
+	queueM, err := client.CreateQueue(ctx, &sqs.CreateQueueInput{
+		QueueName: aws.String(queueName),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	return &queueConfig{
+			Container: &container,
+			Config: &queue.QueueConfig{
+				Client: client,
+				URL:    *queueM.QueueUrl,
+			},
+		}, func() {
+			container.Terminate(ctx)
+		}
+}