From a26b3f8a6a8456ebc704c323bb213caa71316b0a Mon Sep 17 00:00:00 2001 From: Simon Herrmann Date: Mon, 9 Feb 2026 17:26:55 +0100 Subject: [PATCH 1/4] add no timescale option, add tests for manager --- database/dbJob.go | 16 +- database/dbJob_test.go | 44 ++--- database/main_test.go | 2 +- go.mod | 12 +- go.sum | 34 ++-- helper/database.go | 2 + helper/databaseTest.go | 41 +++- helper/database_test.go | 2 +- main_test.go | 2 +- manager/handler/connection.go | 4 +- manager/handler/connection_test.go | 57 ++++++ manager/handler/job.go | 10 +- manager/handler/jobArchive.go | 6 +- manager/handler/jobArchive_test.go | 178 +++++++++++++++++ manager/handler/job_test.go | 255 +++++++++++++++++++++++++ manager/handler/main_test.go | 72 +++++++ manager/handler/managerHandler.go | 4 +- manager/handler/managerHandler_test.go | 29 +++ manager/handler/worker.go | 6 +- manager/handler/worker_test.go | 172 +++++++++++++++++ manager/main_test.go | 72 +++++++ manager/manager.go | 7 +- manager/manager_test.go | 99 ++++++++++ manager/routes.go | 12 +- queuer.go | 4 +- queuerWorker.go | 10 + queuer_test.go | 67 +++++++ sql | 2 +- 28 files changed, 1135 insertions(+), 86 deletions(-) create mode 100644 manager/handler/connection_test.go create mode 100644 manager/handler/jobArchive_test.go create mode 100644 manager/handler/job_test.go create mode 100644 manager/handler/main_test.go create mode 100644 manager/handler/managerHandler_test.go create mode 100644 manager/handler/worker_test.go create mode 100644 manager/main_test.go create mode 100644 manager/manager_test.go diff --git a/database/dbJob.go b/database/dbJob.go index b3eed75..de92dd8 100644 --- a/database/dbJob.go +++ b/database/dbJob.go @@ -42,25 +42,27 @@ type JobDBHandlerFunctions interface { type JobDBHandler struct { db *helper.Database EncryptionKey string + WithTimescale bool } // NewJobDBHandler creates a new instance of JobDBHandler. // It initializes the database connection and optionally drops existing tables. // If withTableDrop is true, it will drop the existing job tables before creating new ones -func NewJobDBHandler(dbConnection *helper.Database, withTableDrop bool, encryptionKey ...string) (*JobDBHandler, error) { +func NewJobDBHandler(dbConnection *helper.Database, dbConfig *helper.DatabaseConfiguration, encryptionKey ...string) (*JobDBHandler, error) { if dbConnection == nil { return nil, helper.NewError("database connection validation", fmt.Errorf("database connection is nil")) } jobDbHandler := &JobDBHandler{ - db: dbConnection, + db: dbConnection, + WithTimescale: dbConfig.WithTimescale, } if len(encryptionKey) > 0 { jobDbHandler.EncryptionKey = encryptionKey[0] } - if withTableDrop { + if dbConfig.WithTableDrop { err := dbConnection.DropFunctionsFromPublicSchema(loadSql.JobFunctions) if err != nil { return nil, helper.NewError("drop job functions", err) @@ -112,7 +114,7 @@ func (r JobDBHandler) CreateTable() error { defer cancel() // Use the SQL init() function to create all tables, triggers, and indexes - _, err := r.db.Instance.ExecContext(ctx, `SELECT init_job();`) + _, err := r.db.Instance.ExecContext(ctx, `SELECT init_job($1);`, r.WithTimescale) if err != nil { log.Panicf("error initializing job tables: %#v", err) } @@ -623,8 +625,9 @@ func (r JobDBHandler) SelectAllJobsBySearch(search string, lastID int, entries i // AddRetentionArchive updates the retention archive settings for the job archive. func (r JobDBHandler) AddRetentionArchive(retention time.Duration) error { _, err := r.db.Instance.Exec( - `SELECT add_retention_archive($1)`, + `SELECT add_retention_archive($1, $2)`, int(retention.Hours()/24), + r.WithTimescale, ) if err != nil { return helper.NewError("exec", err) @@ -636,7 +639,8 @@ func (r JobDBHandler) AddRetentionArchive(retention time.Duration) error { // RemoveRetentionArchive removes the retention archive settings for the job archive. func (r JobDBHandler) RemoveRetentionArchive() error { _, err := r.db.Instance.Exec( - `SELECT remove_retention_archive()`, + `SELECT remove_retention_archive($1)`, + r.WithTimescale, ) if err != nil { return helper.NewError("exec", err) diff --git a/database/dbJob_test.go b/database/dbJob_test.go index 85ed890..cecf09f 100644 --- a/database/dbJob_test.go +++ b/database/dbJob_test.go @@ -22,7 +22,7 @@ func TestJobNewJobDBHandler(t *testing.T) { t.Run("Valid call NewJobDBHandler", func(t *testing.T) { database := helper.NewTestDatabase(dbConfig) - jobDbHandler, err := NewJobDBHandler(database, true) + jobDbHandler, err := NewJobDBHandler(database, dbConfig) assert.NoError(t, err, "Expected NewJobDBHandler to not return an error") require.NotNil(t, jobDbHandler, "Expected NewJobDBHandler to return a non-nil instance") require.NotNil(t, jobDbHandler.db, "Expected NewJobDBHandler to have a non-nil database instance") @@ -37,7 +37,7 @@ func TestJobNewJobDBHandler(t *testing.T) { }) t.Run("Invalid call NewJobDBHandler with nil database", func(t *testing.T) { - _, err := NewJobDBHandler(nil, true) + _, err := NewJobDBHandler(nil, dbConfig) assert.Error(t, err, "Expected error when creating JobDBHandler with nil database") assert.Contains(t, err.Error(), "database connection is nil", "Expected specific error message for nil database connection") }) @@ -51,7 +51,7 @@ func TestJobCheckTableExistance(t *testing.T) { } database := helper.NewTestDatabase(dbConfig) - jobDbHandler, err := NewJobDBHandler(database, true) + jobDbHandler, err := NewJobDBHandler(database, dbConfig) require.NoError(t, err, "Expected NewJobDBHandler to not return an error") exists, err := jobDbHandler.CheckTablesExistance() @@ -67,7 +67,7 @@ func TestJobCreateTable(t *testing.T) { } database := helper.NewTestDatabase(dbConfig) - jobDbHandler, err := NewJobDBHandler(database, true) + jobDbHandler, err := NewJobDBHandler(database, dbConfig) require.NoError(t, err, "Expected NewJobDBHandler to not return an error") err = jobDbHandler.CreateTable() @@ -82,7 +82,7 @@ func TestJobDropTable(t *testing.T) { } database := helper.NewTestDatabase(dbConfig) - jobDbHandler, err := NewJobDBHandler(database, true) + jobDbHandler, err := NewJobDBHandler(database, dbConfig) require.NoError(t, err, "Expected NewJobDBHandler to not return an error") err = jobDbHandler.DropTables() @@ -97,7 +97,7 @@ func TestJobInsertJob(t *testing.T) { } database := helper.NewTestDatabase(dbConfig) - jobDbHandler, err := NewJobDBHandler(database, true) + jobDbHandler, err := NewJobDBHandler(database, dbConfig) require.NoError(t, err, "Expected NewJobDBHandler to not return an error") job, err := model.NewJob("TestTask", nil, nil) @@ -121,7 +121,7 @@ func TestJobInsertJobTx(t *testing.T) { } database := helper.NewTestDatabase(dbConfig) - jobDbHandler, err := NewJobDBHandler(database, true) + jobDbHandler, err := NewJobDBHandler(database, dbConfig) require.NoError(t, err, "Expected NewJobDBHandler to not return an error") job, err := model.NewJob("TestTask", nil, nil) @@ -151,7 +151,7 @@ func TestJobBatchInsertJobs(t *testing.T) { } database := helper.NewTestDatabase(dbConfig) - jobDbHandler, err := NewJobDBHandler(database, true) + jobDbHandler, err := NewJobDBHandler(database, dbConfig) require.NoError(t, err, "Expected NewJobDBHandler to not return an error") t.Run("Successful batch insert jobs", func(t *testing.T) { @@ -249,7 +249,7 @@ func TestJobUpdateJobsInitial(t *testing.T) { require.NoError(t, err, "Expected UpdateWorker to not return an error") // Now we can proceed with the job insertion and update - jobDbHandler, err := NewJobDBHandler(database, true) + jobDbHandler, err := NewJobDBHandler(database, dbConfig) require.NoError(t, err, "Expected NewJobDBHandler to not return an error") job, err := model.NewJob("TestTask", nil, nil) @@ -276,7 +276,7 @@ func TestJobUpdateJobFinal(t *testing.T) { } database := helper.NewTestDatabase(dbConfig) - jobDbHandler, err := NewJobDBHandler(database, true) + jobDbHandler, err := NewJobDBHandler(database, dbConfig) require.NoError(t, err, "Expected NewJobDBHandler to not return an error") job, err := model.NewJob("TestTask", nil, nil) @@ -302,7 +302,7 @@ func TestJobUpdateJobFinalEncrypted(t *testing.T) { } database := helper.NewTestDatabase(dbConfig) - jobDbHandler, err := NewJobDBHandler(database, true, "test-encryption-key") + jobDbHandler, err := NewJobDBHandler(database, dbConfig, "test-encryption-key") require.NoError(t, err, "Expected NewJobDBHandler to not return an error") job, err := model.NewJob("TestTask", nil, nil) @@ -340,7 +340,7 @@ func TestUpdateStaleJobs(t *testing.T) { } database := helper.NewTestDatabase(dbConfig) - jobDbHandler, err := NewJobDBHandler(database, true) + jobDbHandler, err := NewJobDBHandler(database, dbConfig) assert.NoError(t, err, "Expected NewJobDBHandler to not return an error") workerDbHandler, err := NewWorkerDBHandler(database, true) @@ -450,7 +450,7 @@ func TestJobDeleteJob(t *testing.T) { } database := helper.NewTestDatabase(dbConfig) - jobDbHandler, err := NewJobDBHandler(database, true) + jobDbHandler, err := NewJobDBHandler(database, dbConfig) require.NoError(t, err, "Expected NewJobDBHandler to not return an error") job, err := model.NewJob("TestTask", nil, nil) @@ -484,7 +484,7 @@ func TestJobSelectJob(t *testing.T) { } database := helper.NewTestDatabase(dbConfig) - jobDbHandler, err := NewJobDBHandler(database, true) + jobDbHandler, err := NewJobDBHandler(database, dbConfig) require.NoError(t, err, "Expected NewJobDBHandler to not return an error") job, err := model.NewJob("TestTask", nil, nil) @@ -507,7 +507,7 @@ func TestJobSelectJobEncrypted(t *testing.T) { } database := helper.NewTestDatabase(dbConfig) - jobDbHandler, err := NewJobDBHandler(database, true, "test-encryption-key") + jobDbHandler, err := NewJobDBHandler(database, dbConfig, "test-encryption-key") require.NoError(t, err, "Expected NewJobDBHandler to not return an error") job, err := model.NewJob("TestTask", nil, nil) @@ -543,7 +543,7 @@ func TestJobSelectAllJobs(t *testing.T) { database := helper.NewTestDatabase(dbConfig) newJobCount := 5 - jobDbHandler, err := NewJobDBHandler(database, true) + jobDbHandler, err := NewJobDBHandler(database, dbConfig) require.NoError(t, err, "Expected NewJobDBHandler to not return an error") for i := 0; i < newJobCount; i++ { @@ -573,7 +573,7 @@ func TestJobSelectAllJobsEncrypted(t *testing.T) { database := helper.NewTestDatabase(dbConfig) newJobCount := 5 - jobDbHandler, err := NewJobDBHandler(database, true, "test-encryption-key") + jobDbHandler, err := NewJobDBHandler(database, dbConfig, "test-encryption-key") require.NoError(t, err, "Expected NewJobDBHandler to not return an error") expectedResultsByTaskName := make(map[string]model.Parameters) @@ -641,7 +641,7 @@ func TestJobSelectAllJobsByWorkerRID(t *testing.T) { require.NoError(t, err, "Expected UpdateWorker to not return an error") // Insert jobs associated with the worker - jobDbHandler, err := NewJobDBHandler(database, true) + jobDbHandler, err := NewJobDBHandler(database, dbConfig) require.NoError(t, err, "Expected NewJobDBHandler to not return an error") for i := 0; i < newJobCount; i++ { @@ -673,7 +673,7 @@ func TestJobSelectAllJobsBySearch(t *testing.T) { newJobCountSearch := 5 newJobCountOther := 3 - jobDbHandler, err := NewJobDBHandler(database, true) + jobDbHandler, err := NewJobDBHandler(database, dbConfig) require.NoError(t, err, "Expected NewJobDBHandler to not return an error") // Insert multiple jobs with different names @@ -711,7 +711,7 @@ func TestJobSelectJobFromArchive(t *testing.T) { } database := helper.NewTestDatabase(dbConfig) - jobDbHandler, err := NewJobDBHandler(database, true) + jobDbHandler, err := NewJobDBHandler(database, dbConfig) require.NoError(t, err, "Expected NewJobDBHandler to not return an error") job, err := model.NewJob("TestTask", nil, nil) @@ -741,7 +741,7 @@ func TestJobSelectAllJobsFromArchive(t *testing.T) { } database := helper.NewTestDatabase(dbConfig) - jobDbHandler, err := NewJobDBHandler(database, true) + jobDbHandler, err := NewJobDBHandler(database, dbConfig) require.NoError(t, err, "Expected NewJobDBHandler to not return an error") newJobCount := 5 @@ -780,7 +780,7 @@ func TestJobSelectAllJobsFromArchiveBySearch(t *testing.T) { newJobCountSearch := 5 newJobCountOther := 3 - jobDbHandler, err := NewJobDBHandler(database, true) + jobDbHandler, err := NewJobDBHandler(database, dbConfig) require.NoError(t, err, "Expected NewJobDBHandler to not return an error") // Insert multiple jobs with different names diff --git a/database/main_test.go b/database/main_test.go index d26676c..d637709 100644 --- a/database/main_test.go +++ b/database/main_test.go @@ -14,7 +14,7 @@ var dbPort string func TestMain(m *testing.M) { var teardown func(ctx context.Context, opts ...testcontainers.TerminateOption) error var err error - teardown, dbPort, err = helper.MustStartPostgresContainer() + teardown, dbPort, err = helper.MustStartTimescaleContainer() if err != nil { log.Fatalf("error starting postgres container: %v", err) } diff --git a/go.mod b/go.mod index 7cb7e2e..ed143ca 100644 --- a/go.mod +++ b/go.mod @@ -5,6 +5,7 @@ go 1.25.2 require ( github.com/google/uuid v1.6.0 github.com/joho/godotenv v1.5.1 + github.com/labstack/echo/v5 v5.0.3 github.com/lib/pq v1.10.9 github.com/stretchr/testify v1.11.1 github.com/testcontainers/testcontainers-go v0.40.0 @@ -17,14 +18,10 @@ require ( github.com/containerd/errdefs v1.0.0 // indirect github.com/containerd/errdefs/pkg v0.3.0 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect - github.com/labstack/gommon v0.4.2 // indirect github.com/mattn/go-colorable v0.1.14 // indirect github.com/mattn/go-isatty v0.0.20 // indirect github.com/mattn/go-runewidth v0.0.19 // indirect github.com/spf13/pflag v1.0.10 // indirect - github.com/valyala/bytebufferpool v1.0.0 // indirect - github.com/valyala/fasttemplate v1.2.2 // indirect - golang.org/x/net v0.47.0 // indirect golang.org/x/time v0.14.0 // indirect ) @@ -50,7 +47,6 @@ require ( github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.3 // indirect github.com/jedib0t/go-pretty/v6 v6.7.5 github.com/klauspost/compress v1.18.1 // indirect - github.com/labstack/echo/v4 v4.13.4 github.com/lufia/plan9stats v0.0.0-20251013123823-9fd1530e3ec3 // indirect github.com/magiconair/properties v1.8.10 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect @@ -79,9 +75,9 @@ require ( go.opentelemetry.io/otel v1.38.0 // indirect go.opentelemetry.io/otel/metric v1.38.0 // indirect go.opentelemetry.io/otel/trace v1.38.0 // indirect - golang.org/x/crypto v0.45.0 // indirect - golang.org/x/sys v0.38.0 // indirect - golang.org/x/text v0.31.0 // indirect + golang.org/x/crypto v0.47.0 // indirect + golang.org/x/sys v0.40.0 // indirect + golang.org/x/text v0.33.0 // indirect google.golang.org/grpc v1.77.0 // indirect google.golang.org/protobuf v1.36.10 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect diff --git a/go.sum b/go.sum index f8b939a..a426f2e 100644 --- a/go.sum +++ b/go.sum @@ -76,10 +76,8 @@ github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= -github.com/labstack/echo/v4 v4.13.4 h1:oTZZW+T3s9gAu5L8vmzihV7/lkXGZuITzTQkTEhcXEA= -github.com/labstack/echo/v4 v4.13.4/go.mod h1:g63b33BZ5vZzcIUF8AtRH40DrTlXnx4UMC8rBdndmjQ= -github.com/labstack/gommon v0.4.2 h1:F8qTUNXgG1+6WQmqoUWnz8WiEU60mXVVw0P4ht1WRA0= -github.com/labstack/gommon v0.4.2/go.mod h1:QlUFxVM+SNXhDL/Z7YhocGIBYOiwB0mXm1+1bAPHPyU= +github.com/labstack/echo/v5 v5.0.3 h1:Jql8sDtCYXrhh2Mbs6jKwjR6r7X8FSQQmch+w6QS7kc= +github.com/labstack/echo/v5 v5.0.3/go.mod h1:SyvlSdObGjRXeQfCCXW/sybkZdOOQZBmpKF0bvALaeo= github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw= github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o= github.com/lufia/plan9stats v0.0.0-20251013123823-9fd1530e3ec3 h1:PwQumkgq4/acIiZhtifTV5OUqqiP82UAl0h87xj/l9k= @@ -150,10 +148,6 @@ github.com/tklauser/go-sysconf v0.3.16 h1:frioLaCQSsF5Cy1jgRBrzr6t502KIIwQ0MArYI github.com/tklauser/go-sysconf v0.3.16/go.mod h1:/qNL9xxDhc7tx3HSRsLWNnuzbVfh3e7gh/BmM179nYI= github.com/tklauser/numcpus v0.11.0 h1:nSTwhKH5e1dMNsCdVBukSZrURJRoHbSEQjdEbY+9RXw= github.com/tklauser/numcpus v0.11.0/go.mod h1:z+LwcLq54uWZTX0u/bGobaV34u6V7KNlTZejzM6/3MQ= -github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw= -github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc= -github.com/valyala/fasttemplate v1.2.2 h1:lxLXG0uE3Qnshl9QyaK6XJxMXlQZELvChBOCmQD0Loo= -github.com/valyala/fasttemplate v1.2.2/go.mod h1:KHLXt3tVN2HBp8eijSv/kGJopbvo7S+qRAEEKiv+SiQ= github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0= github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= @@ -176,24 +170,24 @@ go.opentelemetry.io/otel/trace v1.38.0 h1:Fxk5bKrDZJUH+AMyyIXGcFAPah0oRcT+LuNtJr go.opentelemetry.io/otel/trace v1.38.0/go.mod h1:j1P9ivuFsTceSWe1oY+EeW3sc+Pp42sO++GHkg4wwhs= go.opentelemetry.io/proto/otlp v1.0.0 h1:T0TX0tmXU8a3CbNXzEKGeU5mIVOdf0oykP+u2lIVU/I= go.opentelemetry.io/proto/otlp v1.0.0/go.mod h1:Sy6pihPLfYHkr3NkUbEhGHFhINUSI/v80hjKIs5JXpM= -golang.org/x/crypto v0.45.0 h1:jMBrvKuj23MTlT0bQEOBcAE0mjg8mK9RXFhRH6nyF3Q= -golang.org/x/crypto v0.45.0/go.mod h1:XTGrrkGJve7CYK7J8PEww4aY7gM3qMCElcJQ8n8JdX4= -golang.org/x/net v0.47.0 h1:Mx+4dIFzqraBXUugkia1OOvlD6LemFo1ALMHjrXDOhY= -golang.org/x/net v0.47.0/go.mod h1:/jNxtkgq5yWUGYkaZGqo27cfGZ1c5Nen03aYrrKpVRU= -golang.org/x/sync v0.18.0 h1:kr88TuHDroi+UVf+0hZnirlk8o8T+4MrK6mr60WkH/I= -golang.org/x/sync v0.18.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= +golang.org/x/crypto v0.47.0 h1:V6e3FRj+n4dbpw86FJ8Fv7XVOql7TEwpHapKoMJ/GO8= +golang.org/x/crypto v0.47.0/go.mod h1:ff3Y9VzzKbwSSEzWqJsJVBnWmRwRSHt/6Op5n9bQc4A= +golang.org/x/net v0.49.0 h1:eeHFmOGUTtaaPSGNmjBKpbng9MulQsJURQUAfUwY++o= +golang.org/x/net v0.49.0/go.mod h1:/ysNB2EvaqvesRkuLAyjI1ycPZlQHM3q01F02UY/MV8= +golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4= +golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.38.0 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc= -golang.org/x/sys v0.38.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= -golang.org/x/term v0.37.0 h1:8EGAD0qCmHYZg6J17DvsMy9/wJ7/D/4pV/wfnld5lTU= -golang.org/x/term v0.37.0/go.mod h1:5pB4lxRNYYVZuTLmy8oR2BH8dflOR+IbTYFD8fi3254= -golang.org/x/text v0.31.0 h1:aC8ghyu4JhP8VojJ2lEHBnochRno1sgL6nEi9WGFGMM= -golang.org/x/text v0.31.0/go.mod h1:tKRAlv61yKIjGGHX/4tP1LTbc13YSec1pxVEWXzfoeM= +golang.org/x/sys v0.40.0 h1:DBZZqJ2Rkml6QMQsZywtnjnnGvHza6BTfYFWY9kjEWQ= +golang.org/x/sys v0.40.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= +golang.org/x/term v0.39.0 h1:RclSuaJf32jOqZz74CkPA9qFuVTX7vhLlpfj/IGWlqY= +golang.org/x/term v0.39.0/go.mod h1:yxzUCTP/U+FzoxfdKmLaA0RV1WgE0VY7hXBwKtY/4ww= +golang.org/x/text v0.33.0 h1:B3njUFyqtHDUI5jMn1YIr5B0IE2U0qck04r6d4KPAxE= +golang.org/x/text v0.33.0/go.mod h1:LuMebE6+rBincTi9+xWTY8TztLzKHc/9C1uBCG27+q8= golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI= golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4= google.golang.org/genproto/googleapis/api v0.0.0-20251022142026-3a174f9686a8 h1:mepRgnBZa07I4TRuomDE4sTIYieg/osKmzIf4USdWS4= diff --git a/helper/database.go b/helper/database.go index 06ef059..4e181c2 100644 --- a/helper/database.go +++ b/helper/database.go @@ -58,6 +58,7 @@ type DatabaseConfiguration struct { Schema string SSLMode string WithTableDrop bool + WithTimescale bool } // NewDatabaseConfiguration creates a new DatabaseConfiguration instance. @@ -73,6 +74,7 @@ func NewDatabaseConfiguration() (*DatabaseConfiguration, error) { Schema: os.Getenv("QUEUER_DB_SCHEMA"), SSLMode: os.Getenv("QUEUER_DB_SSLMODE"), WithTableDrop: os.Getenv("QUEUER_DB_WITH_TABLE_DROP") == "true", + WithTimescale: os.Getenv("QUEUER_DB_WITH_TIMESCALE") != "false", } if len(strings.TrimSpace(config.Host)) == 0 || len(strings.TrimSpace(config.Port)) == 0 || len(strings.TrimSpace(config.Database)) == 0 || len(strings.TrimSpace(config.Username)) == 0 || len(strings.TrimSpace(config.Password)) == 0 || len(strings.TrimSpace(config.Schema)) == 0 { return nil, fmt.Errorf("QUEUER_DB_HOST, QUEUER_DB_PORT, QUEUER_DB_DATABASE, QUEUER_DB_USERNAME, QUEUER_DB_PASSWORD and QUEUER_DB_SCHEMA environment variables must be set") diff --git a/helper/databaseTest.go b/helper/databaseTest.go index 7269db9..f501b0b 100644 --- a/helper/databaseTest.go +++ b/helper/databaseTest.go @@ -19,11 +19,11 @@ const ( dbPwd = "password" ) -// MustStartPostgresContainer starts a PostgreSQL container for testing purposes. -// It uses the timescale/timescaledb image with PostgreSQL 17. +// MustStartTimescaleContainer starts a PostgreSQL container for testing purposes. +// It uses the timescale/timescaledb:latest-pg17 image. // It returns a function to terminate the container, the port on which the database is accessible, // and an error if the container could not be started. -func MustStartPostgresContainer() (func(ctx context.Context, opts ...testcontainers.TerminateOption) error, string, error) { +func MustStartTimescaleContainer() (func(ctx context.Context, opts ...testcontainers.TerminateOption) error, string, error) { ctx := context.Background() pgContainer, err := postgres.Run( @@ -54,6 +54,41 @@ func MustStartPostgresContainer() (func(ctx context.Context, opts ...testcontain return pgContainer.Terminate, u.Port(), err } +// MustStartTimescaleContainer starts a PostgreSQL container for testing purposes. +// It uses the postgres:latest image. +// It returns a function to terminate the container, the port on which the database is accessible, +// and an error if the container could not be started. +func MustStartPostgresContainer() (func(ctx context.Context, opts ...testcontainers.TerminateOption) error, string, error) { + ctx := context.Background() + + pgContainer, err := postgres.Run( + ctx, + "postgres:latest", + postgres.WithDatabase(dbName), + postgres.WithUsername(dbUser), + postgres.WithPassword(dbPwd), + testcontainers.WithWaitStrategy( + wait.ForLog("database system is ready to accept connections"). + WithOccurrence(2).WithStartupTimeout(5*time.Second), + ), + ) + if err != nil { + return nil, "", fmt.Errorf("error starting postgres container: %w", err) + } + + connStr, err := pgContainer.ConnectionString(ctx, "sslmode=disable") + if err != nil { + return nil, "", fmt.Errorf("error getting connection string: %w", err) + } + + u, err := url.Parse(connStr) + if err != nil { + return nil, "", fmt.Errorf("error parsing connection string: %v", err) + } + + return pgContainer.Terminate, u.Port(), err +} + // NewTestDatabase creates a new Database instance for testing purposes. // It initializes the database with the provided configuration and the name "test_db". // It returns a pointer to the new Database instance. diff --git a/helper/database_test.go b/helper/database_test.go index c37ec6c..dfa0232 100644 --- a/helper/database_test.go +++ b/helper/database_test.go @@ -16,7 +16,7 @@ var dbPort string func TestMain(m *testing.M) { var teardown func(ctx context.Context, opts ...testcontainers.TerminateOption) error var err error - teardown, dbPort, err = MustStartPostgresContainer() + teardown, dbPort, err = MustStartTimescaleContainer() if err != nil { log.Fatalf("error starting postgres container: %v", err) } diff --git a/main_test.go b/main_test.go index 4158dac..6b7e6e2 100644 --- a/main_test.go +++ b/main_test.go @@ -14,7 +14,7 @@ var dbPort string func TestMain(m *testing.M) { var teardown func(ctx context.Context, opts ...testcontainers.TerminateOption) error var err error - teardown, dbPort, err = helper.MustStartPostgresContainer() + teardown, dbPort, err = helper.MustStartTimescaleContainer() if err != nil { log.Fatalf("error starting postgres container: %v", err) } diff --git a/manager/handler/connection.go b/manager/handler/connection.go index a7c28b3..d5bdbba 100644 --- a/manager/handler/connection.go +++ b/manager/handler/connection.go @@ -3,11 +3,11 @@ package handler import ( "net/http" - "github.com/labstack/echo/v4" + "github.com/labstack/echo/v5" ) // ListConnections retrieves all active connections -func (m *ManagerHandler) ListConnections(c echo.Context) error { +func (m *ManagerHandler) ListConnections(c *echo.Context) error { connections, err := m.queuer.GetConnections() if err != nil { return c.String(http.StatusInternalServerError, "Failed to retrieve connections") diff --git a/manager/handler/connection_test.go b/manager/handler/connection_test.go new file mode 100644 index 0000000..a37989e --- /dev/null +++ b/manager/handler/connection_test.go @@ -0,0 +1,57 @@ +package handler + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/labstack/echo/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestListConnectionsHandlers(t *testing.T) { + handler := NewManagerHandler(queue) + e := echo.New() + + // Test ListConnections + t.Run("ListConnections returns all active connections", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/connections", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + err := handler.ListConnections(c) + require.NoError(t, err) + assert.Equal(t, http.StatusOK, rec.Code) + + var connections []map[string]interface{} + err = json.Unmarshal(rec.Body.Bytes(), &connections) + require.NoError(t, err) + + // Should have at least the connection from the test setup + assert.GreaterOrEqual(t, len(connections), 0) + + // If there are connections, verify structure contains expected fields + if len(connections) > 0 { + conn := connections[0] + assert.Contains(t, conn, "username") + assert.Contains(t, conn, "database") + } + }) + + t.Run("ListConnections returns valid JSON array", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/connections", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + err := handler.ListConnections(c) + require.NoError(t, err) + assert.Equal(t, http.StatusOK, rec.Code) + + // Verify it's a valid JSON array + var connections []interface{} + err = json.Unmarshal(rec.Body.Bytes(), &connections) + require.NoError(t, err) + }) +} diff --git a/manager/handler/job.go b/manager/handler/job.go index 86adbbe..d483192 100644 --- a/manager/handler/job.go +++ b/manager/handler/job.go @@ -5,12 +5,12 @@ import ( "strconv" "github.com/google/uuid" - "github.com/labstack/echo/v4" + "github.com/labstack/echo/v5" "github.com/siherrmann/queuer/model" ) // AddJob handles the addition of a new job -func (m *ManagerHandler) AddJob(c echo.Context) error { +func (m *ManagerHandler) AddJob(c *echo.Context) error { job := model.Job{} err := c.Bind(&job) if err != nil { @@ -26,7 +26,7 @@ func (m *ManagerHandler) AddJob(c echo.Context) error { } // GetJob retrieves a specific job by RID -func (m *ManagerHandler) GetJob(c echo.Context) error { +func (m *ManagerHandler) GetJob(c *echo.Context) error { ridStr := c.Param("rid") rid, err := uuid.Parse(ridStr) if err != nil { @@ -42,7 +42,7 @@ func (m *ManagerHandler) GetJob(c echo.Context) error { } // ListJobs retrieves a paginated list of jobs -func (m *ManagerHandler) ListJobs(c echo.Context) error { +func (m *ManagerHandler) ListJobs(c *echo.Context) error { lastIdStr := c.QueryParam("lastId") limitStr := c.QueryParam("limit") @@ -75,7 +75,7 @@ func (m *ManagerHandler) ListJobs(c echo.Context) error { } // CancelJob cancels a specific job by RID -func (m *ManagerHandler) CancelJob(c echo.Context) error { +func (m *ManagerHandler) CancelJob(c *echo.Context) error { ridStr := c.Param("rid") rid, err := uuid.Parse(ridStr) if err != nil { diff --git a/manager/handler/jobArchive.go b/manager/handler/jobArchive.go index 56d8753..7435d5e 100644 --- a/manager/handler/jobArchive.go +++ b/manager/handler/jobArchive.go @@ -5,11 +5,11 @@ import ( "strconv" "github.com/google/uuid" - "github.com/labstack/echo/v4" + "github.com/labstack/echo/v5" ) // GetJobArchive retrieves a specific archived job by RID -func (m *ManagerHandler) GetJobArchive(c echo.Context) error { +func (m *ManagerHandler) GetJobArchive(c *echo.Context) error { ridStr := c.Param("rid") rid, err := uuid.Parse(ridStr) if err != nil { @@ -25,7 +25,7 @@ func (m *ManagerHandler) GetJobArchive(c echo.Context) error { } // ListJobArchives retrieves a paginated list of archived jobs -func (m *ManagerHandler) ListJobArchives(c echo.Context) error { +func (m *ManagerHandler) ListJobArchives(c *echo.Context) error { lastIdStr := c.QueryParam("lastId") limitStr := c.QueryParam("limit") diff --git a/manager/handler/jobArchive_test.go b/manager/handler/jobArchive_test.go new file mode 100644 index 0000000..ec28025 --- /dev/null +++ b/manager/handler/jobArchive_test.go @@ -0,0 +1,178 @@ +package handler + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/google/uuid" + "github.com/labstack/echo/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestGetJobArchiveHandler(t *testing.T) { + handler := NewManagerHandler(queue) + e := echo.New() + + // First, create and complete a job so we have an archived job to test with + var archivedJobRID uuid.UUID + t.Run("Setup - Create and complete a job", func(t *testing.T) { + job, err := queue.AddJob("test-task", nil, 1) + require.NoError(t, err) + require.NotNil(t, job) + archivedJobRID = job.RID + + // Wait for the job to be picked up and completed + performedJob := queue.WaitForJobFinished(archivedJobRID, 5*time.Second) + require.NoError(t, err) + require.Equal(t, archivedJobRID, performedJob.RID) + }) + + t.Run("GetJobArchive with valid RID", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/archives/jobs/"+archivedJobRID.String(), nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + c.SetPathValues([]echo.PathValue{{Name: "rid", Value: archivedJobRID.String()}}) + + err := handler.GetJobArchive(c) + require.NoError(t, err) + assert.Equal(t, http.StatusOK, rec.Code) + + var job map[string]interface{} + err = json.Unmarshal(rec.Body.Bytes(), &job) + require.NoError(t, err) + assert.Equal(t, archivedJobRID.String(), job["rid"]) + }) + + t.Run("GetJobArchive with invalid RID format", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/archives/jobs/invalid-uuid", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + c.SetPathValues([]echo.PathValue{{Name: "rid", Value: "invalid-uuid"}}) + err := handler.GetJobArchive(c) + require.NoError(t, err) + assert.Equal(t, http.StatusBadRequest, rec.Code) + assert.Contains(t, rec.Body.String(), "Invalid job archive RID format") + }) + + t.Run("GetJobArchive with non-existent RID", func(t *testing.T) { + nonExistentRID := uuid.New() + req := httptest.NewRequest(http.MethodGet, "/api/v1/archives/jobs/"+nonExistentRID.String(), nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + c.SetPathValues([]echo.PathValue{{Name: "rid", Value: nonExistentRID.String()}}) + + err := handler.GetJobArchive(c) + require.NoError(t, err) + assert.Equal(t, http.StatusNotFound, rec.Code) + assert.Contains(t, rec.Body.String(), "Archived job not found") + }) +} + +func TestListJobArchivesHandler(t *testing.T) { + handler := NewManagerHandler(queue) + e := echo.New() + + t.Run("ListJobArchives with default pagination", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/archives/jobs", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + err := handler.ListJobArchives(c) + require.NoError(t, err) + assert.Equal(t, http.StatusOK, rec.Code) + + var jobs []map[string]interface{} + err = json.Unmarshal(rec.Body.Bytes(), &jobs) + require.NoError(t, err) + assert.GreaterOrEqual(t, len(jobs), 1) + }) + + t.Run("ListJobArchives with custom limit", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/archives/jobs?limit=5", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + err := handler.ListJobArchives(c) + require.NoError(t, err) + assert.Equal(t, http.StatusOK, rec.Code) + + var jobs []map[string]interface{} + err = json.Unmarshal(rec.Body.Bytes(), &jobs) + require.NoError(t, err) + assert.LessOrEqual(t, len(jobs), 5) + }) + + t.Run("ListJobArchives with custom lastId and limit", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/archives/jobs?lastId=0&limit=3", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + err := handler.ListJobArchives(c) + require.NoError(t, err) + assert.Equal(t, http.StatusOK, rec.Code) + + var jobs []map[string]interface{} + err = json.Unmarshal(rec.Body.Bytes(), &jobs) + require.NoError(t, err) + assert.LessOrEqual(t, len(jobs), 3) + }) + + t.Run("ListJobArchives with invalid lastId", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/archives/jobs?lastId=invalid", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + err := handler.ListJobArchives(c) + require.NoError(t, err) + assert.Equal(t, http.StatusBadRequest, rec.Code) + assert.Contains(t, rec.Body.String(), "Invalid lastId format") + }) + + t.Run("ListJobArchives with negative lastId", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/archives/jobs?lastId=-1", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + err := handler.ListJobArchives(c) + require.NoError(t, err) + assert.Equal(t, http.StatusBadRequest, rec.Code) + assert.Contains(t, rec.Body.String(), "Invalid lastId format") + }) + + t.Run("ListJobArchives with invalid limit", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/archives/jobs?limit=invalid", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + err := handler.ListJobArchives(c) + require.NoError(t, err) + assert.Equal(t, http.StatusBadRequest, rec.Code) + assert.Contains(t, rec.Body.String(), "Invalid limit") + }) + + t.Run("ListJobArchives with limit too high", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/archives/jobs?limit=101", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + err := handler.ListJobArchives(c) + require.NoError(t, err) + assert.Equal(t, http.StatusBadRequest, rec.Code) + assert.Contains(t, rec.Body.String(), "Invalid limit") + }) + + t.Run("ListJobArchives with limit zero", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/archives/jobs?limit=0", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + err := handler.ListJobArchives(c) + require.NoError(t, err) + assert.Equal(t, http.StatusBadRequest, rec.Code) + assert.Contains(t, rec.Body.String(), "Invalid limit") + }) +} diff --git a/manager/handler/job_test.go b/manager/handler/job_test.go new file mode 100644 index 0000000..9cfecc1 --- /dev/null +++ b/manager/handler/job_test.go @@ -0,0 +1,255 @@ +package handler + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/google/uuid" + "github.com/labstack/echo/v5" + "github.com/siherrmann/queuer/model" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestAddJobHandler(t *testing.T) { + handler := NewManagerHandler(queue) + e := echo.New() + + t.Run("AddJob with valid data", func(t *testing.T) { + jobJSON := `{"task_name":"TestTask","parameters":[1]}` + req := httptest.NewRequest(http.MethodPost, "/api/v1/jobs", strings.NewReader(jobJSON)) + req.Header.Set(echo.HeaderContentType, echo.MIMEApplicationJSON) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + err := handler.AddJob(c) + require.NoError(t, err) + + assert.Equal(t, http.StatusCreated, rec.Code) + + var job model.Job + err = json.Unmarshal(rec.Body.Bytes(), &job) + require.NoError(t, err) + assert.Equal(t, "TestTask", job.TaskName) + assert.NotEqual(t, uuid.Nil, job.RID) + }) + + t.Run("AddJob with invalid JSON", func(t *testing.T) { + jobJSON := `{"task_name":"TestTask","parameters":invalid}` + req := httptest.NewRequest(http.MethodPost, "/api/v1/jobs", strings.NewReader(jobJSON)) + req.Header.Set(echo.HeaderContentType, echo.MIMEApplicationJSON) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + err := handler.AddJob(c) + require.NoError(t, err) + + assert.Equal(t, http.StatusBadRequest, rec.Code) + assert.Contains(t, rec.Body.String(), "Invalid job data") + }) + + t.Run("AddJob with non-existent task", func(t *testing.T) { + jobJSON := `{"task_name":"NonExistentTask","parameters":[]}` + req := httptest.NewRequest(http.MethodPost, "/api/v1/jobs", strings.NewReader(jobJSON)) + req.Header.Set(echo.HeaderContentType, echo.MIMEApplicationJSON) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + err := handler.AddJob(c) + require.NoError(t, err) + + // Queuer allows adding jobs with any task name - validation happens at execution time + assert.Equal(t, http.StatusCreated, rec.Code) + + var job model.Job + err = json.Unmarshal(rec.Body.Bytes(), &job) + require.NoError(t, err) + assert.Equal(t, "NonExistentTask", job.TaskName) + }) +} + +func TestGetJobHandler(t *testing.T) { + handler := NewManagerHandler(queue) + e := echo.New() + + t.Run("GetJob with valid RID", func(t *testing.T) { + // First create a job + job, err := queue.AddJob("TestTask", nil, 1) + require.NoError(t, err) + + // Wait a bit for job to be queued + time.Sleep(100 * time.Millisecond) + + req := httptest.NewRequest(http.MethodGet, "/api/v1/jobs/"+job.RID.String(), nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + c.SetPathValues([]echo.PathValue{{Name: "rid", Value: job.RID.String()}}) + + err = handler.GetJob(c) + require.NoError(t, err) + + assert.Equal(t, http.StatusOK, rec.Code) + + var fetchedJob model.Job + err = json.Unmarshal(rec.Body.Bytes(), &fetchedJob) + require.NoError(t, err) + assert.Equal(t, job.RID, fetchedJob.RID) + assert.Equal(t, "TestTask", fetchedJob.TaskName) + }) + + t.Run("GetJob with invalid RID format", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/jobs/invalid-uuid", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + c.SetPathValues([]echo.PathValue{{Name: "rid", Value: "invalid-uuid"}}) + + err := handler.GetJob(c) + require.NoError(t, err) + + assert.Equal(t, http.StatusBadRequest, rec.Code) + assert.Contains(t, rec.Body.String(), "Invalid job RID format") + }) + + t.Run("GetJob with non-existent RID", func(t *testing.T) { + nonExistentRID := uuid.New() + req := httptest.NewRequest(http.MethodGet, "/api/v1/jobs/"+nonExistentRID.String(), nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + c.SetPathValues([]echo.PathValue{{Name: "rid", Value: nonExistentRID.String()}}) + + err := handler.GetJob(c) + require.NoError(t, err) + + assert.Equal(t, http.StatusNotFound, rec.Code) + assert.Contains(t, rec.Body.String(), "Job not found") + }) +} + +func TestListJobsHandler(t *testing.T) { + handler := NewManagerHandler(queue) + e := echo.New() + + t.Run("ListJobs with default pagination", func(t *testing.T) { + // Create multiple jobs + for i := 0; i < 5; i++ { + _, err := queue.AddJob("TestTask", nil, 1) + require.NoError(t, err) + } + + time.Sleep(100 * time.Millisecond) + + req := httptest.NewRequest(http.MethodGet, "/api/v1/jobs", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + err := handler.ListJobs(c) + require.NoError(t, err) + + assert.Equal(t, http.StatusOK, rec.Code) + + var jobs []*model.Job + err = json.Unmarshal(rec.Body.Bytes(), &jobs) + require.NoError(t, err) + assert.GreaterOrEqual(t, len(jobs), 1) + assert.LessOrEqual(t, len(jobs), 10) // Default limit + }) + + t.Run("ListJobs with custom pagination", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/jobs?lastId=0&limit=3", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + err := handler.ListJobs(c) + require.NoError(t, err) + + assert.Equal(t, http.StatusOK, rec.Code) + + var jobs []*model.Job + err = json.Unmarshal(rec.Body.Bytes(), &jobs) + require.NoError(t, err) + assert.LessOrEqual(t, len(jobs), 3) + }) + + t.Run("ListJobs with invalid lastId", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/jobs?lastId=invalid", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + err := handler.ListJobs(c) + require.NoError(t, err) + + assert.Equal(t, http.StatusBadRequest, rec.Code) + assert.Contains(t, rec.Body.String(), "Invalid lastId") + }) + + t.Run("ListJobs with invalid limit", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/jobs?limit=200", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + err := handler.ListJobs(c) + require.NoError(t, err) + + assert.Equal(t, http.StatusBadRequest, rec.Code) + assert.Contains(t, rec.Body.String(), "Invalid limit") + }) +} + +func TestCancelJobHandler(t *testing.T) { + handler := NewManagerHandler(queue) + e := echo.New() + + t.Run("CancelJob with valid RID", func(t *testing.T) { + // Create a job + job, err := queue.AddJob("TestTask", nil, 10) // Long running + require.NoError(t, err) + + time.Sleep(100 * time.Millisecond) + + req := httptest.NewRequest(http.MethodDelete, "/api/v1/jobs/"+job.RID.String(), nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + c.SetPathValues([]echo.PathValue{{Name: "rid", Value: job.RID.String()}}) + + err = handler.CancelJob(c) + require.NoError(t, err) + + assert.Equal(t, http.StatusOK, rec.Code) + + var cancelledJob model.Job + err = json.Unmarshal(rec.Body.Bytes(), &cancelledJob) + require.NoError(t, err) + assert.Equal(t, job.RID, cancelledJob.RID) + }) + + t.Run("CancelJob with invalid RID format", func(t *testing.T) { + req := httptest.NewRequest(http.MethodDelete, "/api/v1/jobs/invalid-uuid", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + c.SetPathValues([]echo.PathValue{{Name: "rid", Value: "invalid-uuid"}}) + + err := handler.CancelJob(c) + require.NoError(t, err) + + assert.Equal(t, http.StatusBadRequest, rec.Code) + assert.Contains(t, rec.Body.String(), "Invalid job RID format") + }) + + t.Run("CancelJob with non-existent RID", func(t *testing.T) { + nonExistentRID := uuid.New() + req := httptest.NewRequest(http.MethodDelete, "/api/v1/jobs/"+nonExistentRID.String(), nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + c.SetPathValues([]echo.PathValue{{Name: "rid", Value: nonExistentRID.String()}}) + + err := handler.CancelJob(c) + require.NoError(t, err) + + assert.Equal(t, http.StatusInternalServerError, rec.Code) + assert.Contains(t, rec.Body.String(), "Failed to cancel job") + }) +} diff --git a/manager/handler/main_test.go b/manager/handler/main_test.go new file mode 100644 index 0000000..5f5cb96 --- /dev/null +++ b/manager/handler/main_test.go @@ -0,0 +1,72 @@ +package handler + +import ( + "context" + "errors" + "log" + "testing" + "time" + + "github.com/siherrmann/queuer" + "github.com/siherrmann/queuer/helper" + "github.com/testcontainers/testcontainers-go" +) + +var dbPort string +var queue *queuer.Queuer + +func testTask(timeSeconds int) error { + time.Sleep(time.Duration(timeSeconds) * time.Second) + return nil +} + +func testTaskFailing() error { + return errors.New("task failed") +} + +func TestMain(m *testing.M) { + var teardown func(ctx context.Context, opts ...testcontainers.TerminateOption) error + var err error + teardown, dbPort, err = helper.MustStartTimescaleContainer() + if err != nil { + log.Fatalf("error starting postgres container: %v", err) + } + + dbConf := &helper.DatabaseConfiguration{ + Host: "localhost", + Port: dbPort, + Database: "database", + Username: "user", + Password: "password", + Schema: "public", + SSLMode: "disable", + WithTableDrop: true, + } + + queue = queuer.NewQueuerWithDB("TestQueuer", 10, "", dbConf) + queue.AddTaskWithName(testTask, "test-task") + queue.AddTaskWithName(testTaskFailing, "test-task-failing") + + ctx, cancel := context.WithCancel(context.Background()) + queue.Start(ctx, cancel) + + // Give queuer time to start + time.Sleep(500 * time.Millisecond) + + exitCode := m.Run() + + // Stop the queuer + cancel() + queue.Stop() + + if teardown != nil { + if err := teardown(context.Background()); err != nil { + log.Fatalf("error tearing down postgres container: %v", err) + } + } + + // Exit with the test exit code + if exitCode != 0 { + log.Fatalf("tests failed with exit code: %d", exitCode) + } +} diff --git a/manager/handler/managerHandler.go b/manager/handler/managerHandler.go index f22d92b..59a2f1c 100644 --- a/manager/handler/managerHandler.go +++ b/manager/handler/managerHandler.go @@ -3,7 +3,7 @@ package handler import ( "net/http" - "github.com/labstack/echo/v4" + "github.com/labstack/echo/v5" "github.com/siherrmann/queuer" ) @@ -18,7 +18,7 @@ func NewManagerHandler(queuer *queuer.Queuer) *ManagerHandler { } // Health check handler -func (m *ManagerHandler) HealthCheck(c echo.Context) error { +func (m *ManagerHandler) HealthCheck(c *echo.Context) error { return c.JSON(http.StatusOK, map[string]string{ "status": "healthy", "service": "queuer-manager", diff --git a/manager/handler/managerHandler_test.go b/manager/handler/managerHandler_test.go new file mode 100644 index 0000000..dbb2fab --- /dev/null +++ b/manager/handler/managerHandler_test.go @@ -0,0 +1,29 @@ +package handler + +import ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/labstack/echo/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestHealthCheck(t *testing.T) { + handler := NewManagerHandler(queue) + e := echo.New() + + t.Run("Should return healthy status", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/health", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + err := handler.HealthCheck(c) + require.NoError(t, err) + + assert.Equal(t, http.StatusOK, rec.Code) + assert.Contains(t, rec.Body.String(), "healthy") + assert.Contains(t, rec.Body.String(), "queuer-manager") + }) +} diff --git a/manager/handler/worker.go b/manager/handler/worker.go index d77293b..3edc035 100644 --- a/manager/handler/worker.go +++ b/manager/handler/worker.go @@ -5,11 +5,11 @@ import ( "strconv" "github.com/google/uuid" - "github.com/labstack/echo/v4" + "github.com/labstack/echo/v5" ) // GetWorker retrieves a specific worker by RID -func (m *ManagerHandler) GetWorker(c echo.Context) error { +func (m *ManagerHandler) GetWorker(c *echo.Context) error { ridStr := c.Param("rid") rid, err := uuid.Parse(ridStr) if err != nil { @@ -25,7 +25,7 @@ func (m *ManagerHandler) GetWorker(c echo.Context) error { } // ListWorkers retrieves a paginated list of workers -func (m *ManagerHandler) ListWorkers(c echo.Context) error { +func (m *ManagerHandler) ListWorkers(c *echo.Context) error { lastIdStr := c.QueryParam("lastId") limitStr := c.QueryParam("limit") diff --git a/manager/handler/worker_test.go b/manager/handler/worker_test.go new file mode 100644 index 0000000..68c5b2b --- /dev/null +++ b/manager/handler/worker_test.go @@ -0,0 +1,172 @@ +package handler + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/google/uuid" + "github.com/labstack/echo/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestGetWorkerHandler(t *testing.T) { + handler := NewManagerHandler(queue) + e := echo.New() + + // Get the queuer's own worker RID + var workerRID uuid.UUID + t.Run("Setup - Get a worker RID", func(t *testing.T) { + workerRid := queue.GetCurrentWorkerRID() + require.NotEqual(t, uuid.Nil, workerRid) + workerRID = workerRid + }) + + t.Run("GetWorker with valid RID", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/workers/"+workerRID.String(), nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + c.SetPathValues([]echo.PathValue{{Name: "rid", Value: workerRID.String()}}) + + err := handler.GetWorker(c) + require.NoError(t, err) + assert.Equal(t, http.StatusOK, rec.Code) + + var worker map[string]interface{} + err = json.Unmarshal(rec.Body.Bytes(), &worker) + require.NoError(t, err) + assert.Equal(t, workerRID.String(), worker["rid"]) + }) + + t.Run("GetWorker with invalid RID format", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/workers/invalid-uuid", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + c.SetPathValues([]echo.PathValue{{Name: "rid", Value: "invalid-uuid"}}) + + err := handler.GetWorker(c) + require.NoError(t, err) + assert.Equal(t, http.StatusBadRequest, rec.Code) + assert.Contains(t, rec.Body.String(), "Invalid worker RID format") + }) + + t.Run("GetWorker with non-existent RID", func(t *testing.T) { + nonExistentRID := uuid.New() + req := httptest.NewRequest(http.MethodGet, "/api/v1/workers/"+nonExistentRID.String(), nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + c.SetPathValues([]echo.PathValue{{Name: "rid", Value: nonExistentRID.String()}}) + + err := handler.GetWorker(c) + require.NoError(t, err) + assert.Equal(t, http.StatusNotFound, rec.Code) + assert.Contains(t, rec.Body.String(), "Worker not found") + }) +} + +func TestListWorkersHandler(t *testing.T) { + handler := NewManagerHandler(queue) + e := echo.New() + + t.Run("ListWorkers with default pagination", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/workers", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + err := handler.ListWorkers(c) + require.NoError(t, err) + assert.Equal(t, http.StatusOK, rec.Code) + + var workers []map[string]interface{} + err = json.Unmarshal(rec.Body.Bytes(), &workers) + require.NoError(t, err) + assert.GreaterOrEqual(t, len(workers), 1) + }) + + t.Run("ListWorkers with custom limit", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/workers?limit=5", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + err := handler.ListWorkers(c) + require.NoError(t, err) + assert.Equal(t, http.StatusOK, rec.Code) + + var workers []map[string]interface{} + err = json.Unmarshal(rec.Body.Bytes(), &workers) + require.NoError(t, err) + assert.LessOrEqual(t, len(workers), 5) + }) + + t.Run("ListWorkers with custom lastId and limit", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/workers?lastId=0&limit=3", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + err := handler.ListWorkers(c) + require.NoError(t, err) + assert.Equal(t, http.StatusOK, rec.Code) + + var workers []map[string]interface{} + err = json.Unmarshal(rec.Body.Bytes(), &workers) + require.NoError(t, err) + assert.LessOrEqual(t, len(workers), 3) + }) + + t.Run("ListWorkers with invalid lastId", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/workers?lastId=invalid", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + err := handler.ListWorkers(c) + require.NoError(t, err) + assert.Equal(t, http.StatusBadRequest, rec.Code) + assert.Contains(t, rec.Body.String(), "Invalid lastId format") + }) + + t.Run("ListWorkers with negative lastId", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/workers?lastId=-1", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + err := handler.ListWorkers(c) + require.NoError(t, err) + assert.Equal(t, http.StatusBadRequest, rec.Code) + assert.Contains(t, rec.Body.String(), "Invalid lastId format") + }) + + t.Run("ListWorkers with invalid limit", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/workers?limit=invalid", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + err := handler.ListWorkers(c) + require.NoError(t, err) + assert.Equal(t, http.StatusBadRequest, rec.Code) + assert.Contains(t, rec.Body.String(), "Invalid limit") + }) + + t.Run("ListWorkers with limit too high", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/workers?limit=101", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + err := handler.ListWorkers(c) + require.NoError(t, err) + assert.Equal(t, http.StatusBadRequest, rec.Code) + assert.Contains(t, rec.Body.String(), "Invalid limit") + }) + + t.Run("ListWorkers with limit zero", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/workers?limit=0", nil) + rec := httptest.NewRecorder() + c := e.NewContext(req, rec) + + err := handler.ListWorkers(c) + require.NoError(t, err) + assert.Equal(t, http.StatusBadRequest, rec.Code) + assert.Contains(t, rec.Body.String(), "Invalid limit") + }) +} diff --git a/manager/main_test.go b/manager/main_test.go new file mode 100644 index 0000000..51ad1ed --- /dev/null +++ b/manager/main_test.go @@ -0,0 +1,72 @@ +package manager + +import ( + "context" + "errors" + "log" + "testing" + "time" + + "github.com/siherrmann/queuer" + "github.com/siherrmann/queuer/helper" + "github.com/testcontainers/testcontainers-go" +) + +var dbPort string +var queue *queuer.Queuer + +func testTask(timeSeconds int) error { + time.Sleep(time.Duration(timeSeconds) * time.Second) + return nil +} + +func testTaskFailing() error { + return errors.New("task failed") +} + +func TestMain(m *testing.M) { + var teardown func(ctx context.Context, opts ...testcontainers.TerminateOption) error + var err error + teardown, dbPort, err = helper.MustStartTimescaleContainer() + if err != nil { + log.Fatalf("error starting postgres container: %v", err) + } + + dbConf := &helper.DatabaseConfiguration{ + Host: "localhost", + Port: dbPort, + Database: "database", + Username: "user", + Password: "password", + Schema: "public", + SSLMode: "disable", + WithTableDrop: true, + } + + queue = queuer.NewQueuerWithDB("TestQueuer", 10, "", dbConf) + queue.AddTaskWithName(testTask, "test-task") + queue.AddTaskWithName(testTaskFailing, "test-task-failing") + + ctx, cancel := context.WithCancel(context.Background()) + queue.Start(ctx, cancel) + + // Give queuer time to start + time.Sleep(500 * time.Millisecond) + + exitCode := m.Run() + + // Stop the queuer + cancel() + queue.Stop() + + if teardown != nil { + if err := teardown(context.Background()); err != nil { + log.Fatalf("error tearing down postgres container: %v", err) + } + } + + // Exit with the test exit code + if exitCode != 0 { + log.Fatalf("tests failed with exit code: %d", exitCode) + } +} diff --git a/manager/manager.go b/manager/manager.go index 3a60243..a95a09e 100644 --- a/manager/manager.go +++ b/manager/manager.go @@ -3,7 +3,7 @@ package manager import ( "strconv" - "github.com/labstack/echo/v4" + "github.com/labstack/echo/v5" "github.com/siherrmann/queuer" ) @@ -13,5 +13,8 @@ func ManagerServer(port int, maxConcurrency int) { e := echo.New() SetupRoutes(e, queuerInstance) - e.Logger.Fatal(e.Start(":" + strconv.Itoa(port))) + err := e.Start(":" + strconv.Itoa(port)) + if err != nil { + panic(err) + } } diff --git a/manager/manager_test.go b/manager/manager_test.go new file mode 100644 index 0000000..7c72f30 --- /dev/null +++ b/manager/manager_test.go @@ -0,0 +1,99 @@ +package manager + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/labstack/echo/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestSetupRoutes(t *testing.T) { + e := echo.New() + SetupRoutes(e, queue) + + t.Run("Health endpoint is registered", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/health", nil) + rec := httptest.NewRecorder() + e.ServeHTTP(rec, req) + + assert.Equal(t, http.StatusOK, rec.Code) + assert.Contains(t, rec.Body.String(), "healthy") + }) + + t.Run("Root endpoint is registered", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/", nil) + rec := httptest.NewRecorder() + e.ServeHTTP(rec, req) + + assert.Equal(t, http.StatusOK, rec.Code) + assert.Contains(t, rec.Body.String(), "healthy") + }) + + t.Run("Jobs endpoints are registered", func(t *testing.T) { + // POST /api/v1/jobs + req := httptest.NewRequest(http.MethodPost, "/api/v1/jobs", nil) + req.Header.Set("Content-Type", "application/json") + rec := httptest.NewRecorder() + e.ServeHTTP(rec, req) + // Should not be 404 + assert.NotEqual(t, http.StatusNotFound, rec.Code) + + // GET /api/v1/jobs + req = httptest.NewRequest(http.MethodGet, "/api/v1/jobs", nil) + rec = httptest.NewRecorder() + e.ServeHTTP(rec, req) + assert.Equal(t, http.StatusOK, rec.Code) + }) + + t.Run("Job archives endpoints are registered", func(t *testing.T) { + // GET /api/v1/job-archives + req := httptest.NewRequest(http.MethodGet, "/api/v1/job-archives", nil) + rec := httptest.NewRecorder() + e.ServeHTTP(rec, req) + assert.Equal(t, http.StatusOK, rec.Code) + }) + + t.Run("Workers endpoints are registered", func(t *testing.T) { + // GET /api/v1/workers + req := httptest.NewRequest(http.MethodGet, "/api/v1/workers", nil) + rec := httptest.NewRecorder() + e.ServeHTTP(rec, req) + assert.Equal(t, http.StatusOK, rec.Code) + + var workers []map[string]interface{} + err := json.Unmarshal(rec.Body.Bytes(), &workers) + require.NoError(t, err) + assert.GreaterOrEqual(t, len(workers), 1) + }) + + t.Run("Connections endpoint is registered", func(t *testing.T) { + // GET /api/v1/connections + req := httptest.NewRequest(http.MethodGet, "/api/v1/connections", nil) + rec := httptest.NewRecorder() + e.ServeHTTP(rec, req) + assert.Equal(t, http.StatusOK, rec.Code) + }) + + t.Run("CORS middleware is enabled", func(t *testing.T) { + req := httptest.NewRequest(http.MethodOptions, "/api/v1/workers", nil) + req.Header.Set("Origin", "http://example.com") + req.Header.Set("Access-Control-Request-Method", "GET") + rec := httptest.NewRecorder() + e.ServeHTTP(rec, req) + + // CORS should add these headers + assert.NotEmpty(t, rec.Header().Get("Access-Control-Allow-Origin")) + }) + + t.Run("Non-existent endpoint returns 404", func(t *testing.T) { + req := httptest.NewRequest(http.MethodGet, "/api/v1/nonexistent", nil) + rec := httptest.NewRecorder() + e.ServeHTTP(rec, req) + + assert.Equal(t, http.StatusNotFound, rec.Code) + }) +} diff --git a/manager/routes.go b/manager/routes.go index 0383465..4813bc5 100644 --- a/manager/routes.go +++ b/manager/routes.go @@ -1,8 +1,10 @@ package manager import ( - "github.com/labstack/echo/v4" - "github.com/labstack/echo/v4/middleware" + "net/http" + + "github.com/labstack/echo/v5" + "github.com/labstack/echo/v5/middleware" "github.com/siherrmann/queuer" "github.com/siherrmann/queuer/manager/handler" ) @@ -12,9 +14,11 @@ func SetupRoutes(e *echo.Echo, queuerInstance *queuer.Queuer) { h := handler.NewManagerHandler(queuerInstance) // Middleware - e.Use(middleware.Logger()) e.Use(middleware.Recover()) - e.Use(middleware.CORS()) + e.Use(middleware.CORSWithConfig(middleware.CORSConfig{ + AllowOrigins: []string{"*"}, + AllowMethods: []string{http.MethodGet, http.MethodPost, http.MethodPut, http.MethodDelete}, + })) e.GET("/health", h.HealthCheck) e.GET("/", h.HealthCheck) diff --git a/queuer.go b/queuer.go index 41f11b1..73c9cd2 100644 --- a/queuer.go +++ b/queuer.go @@ -111,7 +111,7 @@ func NewQueuerWithDB(name string, maxConcurrency int, encryptionKey string, dbCo } // DBs - dbJob, err := database.NewJobDBHandler(dbCon, dbConfig.WithTableDrop) + dbJob, err := database.NewJobDBHandler(dbCon, dbConfig) if err != nil { log.Panicf("error creating job db handler: %s", err.Error()) } @@ -199,7 +199,7 @@ func NewStaticQueuer(logLevel slog.Leveler, dbConfig *helper.DatabaseConfigurati } // DBs - dbJob, err := database.NewJobDBHandler(dbCon, dbConfig.WithTableDrop) + dbJob, err := database.NewJobDBHandler(dbCon, dbConfig) if err != nil { log.Panicf("error creating job db handler: %s", err.Error()) } diff --git a/queuerWorker.go b/queuerWorker.go index ad447af..e16c63c 100644 --- a/queuerWorker.go +++ b/queuerWorker.go @@ -8,6 +8,16 @@ import ( "github.com/siherrmann/queuer/model" ) +func (q *Queuer) GetCurrentWorkerRID() uuid.UUID { + q.workerMu.RLock() + defer q.workerMu.RUnlock() + + if q.worker != nil { + return q.worker.RID + } + return uuid.Nil +} + // StopWorkerGracefully sets the status of the specified worker to 'STOPPING' // to cancel running jobs when stopping. func (q *Queuer) StopWorker(workerRid uuid.UUID) error { diff --git a/queuer_test.go b/queuer_test.go index 1771e82..91d19c6 100644 --- a/queuer_test.go +++ b/queuer_test.go @@ -851,3 +851,70 @@ func TestStopWorkerGracefullyWithRunningJobs(t *testing.T) { // Note: Both queuers will be stopped automatically by their heartbeat tickers }) } + +func TestJobArchiveRetention(t *testing.T) { + t.Run("TimescaleDB retention policy", func(t *testing.T) { + teardown, timescalePort, err := helper.MustStartTimescaleContainer() + require.NoError(t, err) + defer teardown(context.Background()) + + queuer := NewQueuerWithDB("test-retention", 10, "", &helper.DatabaseConfiguration{ + Host: "localhost", Port: timescalePort, Database: "database", + Username: "user", Password: "password", Schema: "public", + SSLMode: "disable", WithTableDrop: true, WithTimescale: true, + }) + queuer.AddTaskWithName(func() (string, error) { return "success", nil }, "test-task") + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + queuer.Start(ctx, cancel, &model.MasterSettings{JobDeleteThreshold: 1 * time.Hour}) + time.Sleep(6 * time.Second) + + master, err := queuer.dbMaster.SelectMaster() + require.NoError(t, err) + assert.Equal(t, 1*time.Hour, master.Settings.JobDeleteThreshold) + + queuer.Stop() + }) + + t.Run("PostgreSQL trigger-based retention", func(t *testing.T) { + teardown, postgresPort, err := helper.MustStartPostgresContainer() + require.NoError(t, err) + defer teardown(context.Background()) + + queuer := NewQueuerWithDB("test-retention", 10, "", &helper.DatabaseConfiguration{ + Host: "localhost", Port: postgresPort, Database: "database", + Username: "user", Password: "password", Schema: "public", + SSLMode: "disable", WithTableDrop: true, WithTimescale: false, + }) + queuer.AddTaskWithName(func() (string, error) { return "success", nil }, "test-task") + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + queuer.Start(ctx, cancel, &model.MasterSettings{JobDeleteThreshold: 1 * time.Hour}) + time.Sleep(6 * time.Second) + + // Insert old job + oldJobRID := uuid.New() + _, err = queuer.DB.Exec(`INSERT INTO job_archive (id, rid, worker_id, worker_rid, task_name, status, updated_at, created_at) + VALUES (1, $1, 1, $2, 'test-task', 'SUCCEEDED', CURRENT_TIMESTAMP - INTERVAL '2 hours', CURRENT_TIMESTAMP - INTERVAL '2 hours')`, + oldJobRID, queuer.worker.RID) + require.NoError(t, err) + + // Add 15 jobs to trigger cleanup (trigger runs every 10th insert) + for i := 0; i < 15; i++ { + _, err := queuer.AddJob("test-task", nil) + require.NoError(t, err) + } + + // Wait for jobs to complete and trigger to execute + time.Sleep(5 * time.Second) + + // Verify old job was deleted + deletedJob, err := queuer.dbJob.SelectJobFromArchive(oldJobRID) + assert.Error(t, err) + assert.Nil(t, deletedJob) + + queuer.Stop() + }) +} diff --git a/sql b/sql index f7b8a69..f0fc93d 160000 --- a/sql +++ b/sql @@ -1 +1 @@ -Subproject commit f7b8a69559108ee03c5ee61dcb4b894bd1c76bf6 +Subproject commit f0fc93dd45b988d91da88ccacbab6eb8a78f8a74 From 1e8eb8cf40069cabe112b8506a82761acc8c88bf Mon Sep 17 00:00:00 2001 From: Simon Herrmann Date: Sat, 11 Jul 2026 16:36:40 +0200 Subject: [PATCH 2/4] add parent rid functionality, add tests, add context to queuer --- core/runner.go | 51 ++++++++++++++++++++++++------- core/runner_test.go | 40 ++++++++++++++++++++++++ database/dbJob_test.go | 45 +++++++++++++++++++++++++++ helper/database.go | 3 ++ helper/database_test.go | 35 +++++++++++++++++++++ helper/prettyLog_test.go | 58 +++++++++++++++++++++++++++++++++++ helper/task.go | 19 +++++++++--- helper/task_test.go | 12 ++++---- manager/manager_test.go | 3 ++ model/options.go | 7 +++-- queuerJob.go | 62 +++++++++++++++++++++++++++++++++++++ queuerJobCtx_test.go | 66 ++++++++++++++++++++++++++++++++++++++++ queuer_test.go | 8 ----- sql | 2 +- 14 files changed, 378 insertions(+), 33 deletions(-) create mode 100644 helper/prettyLog_test.go create mode 100644 queuerJobCtx_test.go diff --git a/core/runner.go b/core/runner.go index d8c5b97..e5d9635 100644 --- a/core/runner.go +++ b/core/runner.go @@ -8,17 +8,22 @@ import ( "sync" "time" + "github.com/google/uuid" "github.com/siherrmann/queuer/helper" "github.com/siherrmann/queuer/model" vh "github.com/siherrmann/validator/helper" ) +type ContextKey string +const JobRIDContextKey ContextKey = "JobRIDContextKey" + type Runner struct { - cancel context.CancelFunc - cancelMu sync.RWMutex - Options *model.Options - Task interface{} - Parameters model.Parameters + cancel context.CancelFunc + cancelMu sync.RWMutex + Options *model.Options + JobRID *uuid.UUID + Task interface{} + Parameters model.Parameters // Result channel to return results ResultsChannel chan []interface{} ErrorChannel chan error @@ -31,14 +36,22 @@ func NewRunner(options *model.Options, task interface{}, parameters ...interface taskInputParameters, err := helper.GetInputParametersFromTask(task) if err != nil { return nil, helper.NewError("getting task input parameters", err) - } else if len(taskInputParameters) != len(parameters) { - return nil, fmt.Errorf("task expects %d parameters, got %d", len(taskInputParameters), len(parameters)) + } + + startIndex := 0 + contextType := reflect.TypeOf((*context.Context)(nil)).Elem() + if len(taskInputParameters) > 0 && taskInputParameters[0] == contextType { + startIndex = 1 + } + + if len(taskInputParameters)-startIndex != len(parameters) { + return nil, fmt.Errorf("task expects %d parameters, got %d", len(taskInputParameters)-startIndex, len(parameters)) } for i, param := range parameters { - paramConverted, err := vh.AnyToType(param, taskInputParameters[i]) + paramConverted, err := vh.AnyToType(param, taskInputParameters[i+startIndex]) if err != nil { - return nil, fmt.Errorf("error converting parameter %d to type %s: %v", i, taskInputParameters[i].Kind(), err) + return nil, fmt.Errorf("error converting parameter %d to type %s: %v", i, taskInputParameters[i+startIndex].Kind(), err) } parameters[i] = paramConverted } @@ -73,6 +86,7 @@ func NewRunnerFromJob(task *model.Task, job *model.Job) (*Runner, error) { return nil, fmt.Errorf("error creating runner from job: %v", err) } + runner.JobRID = &job.RID return runner, nil } @@ -111,7 +125,21 @@ func (r *Runner) Run(ctx context.Context) { }() taskFunc := reflect.ValueOf(r.Task) - results := taskFunc.Call(r.Parameters.ToReflectValues()) + + var callParameters []reflect.Value + taskType := reflect.TypeOf(r.Task) + contextType := reflect.TypeOf((*context.Context)(nil)).Elem() + + if taskType.NumIn() > 0 && taskType.In(0) == contextType { + jobCtx := ctx + if r.JobRID != nil { + jobCtx = context.WithValue(ctx, JobRIDContextKey, *r.JobRID) + } + callParameters = append(callParameters, reflect.ValueOf(jobCtx)) + } + + callParameters = append(callParameters, r.Parameters.ToReflectValues()...) + results := taskFunc.Call(callParameters) resultValues := []interface{}{} for _, result := range results { resultValues = append(resultValues, result.Interface()) @@ -125,7 +153,8 @@ func (r *Runner) Run(ctx context.Context) { var ok bool if len(resultValues) > 0 { - if err, ok = resultValues[len(resultValues)-1].(error); ok || (outputParameters[1].String() == "error" && resultValues[len(resultValues)-1] == nil) { + lastOutIdx := len(outputParameters) - 1 + if err, ok = resultValues[len(resultValues)-1].(error); ok || (lastOutIdx >= 0 && outputParameters[lastOutIdx].String() == "error" && resultValues[len(resultValues)-1] == nil) { resultValues = resultValues[:len(resultValues)-1] } } diff --git a/core/runner_test.go b/core/runner_test.go index 03f1ecb..5f06199 100644 --- a/core/runner_test.go +++ b/core/runner_test.go @@ -692,3 +692,43 @@ func TestNewRunnerWithNestedStruct(t *testing.T) { }) } } + +func TestRunnerContextInjection(t *testing.T) { + rid := uuid.New() + taskFn := func(ctx context.Context, p string) string { + ctxRid := ctx.Value(JobRIDContextKey) + if ctxRid != nil { + if id, ok := ctxRid.(uuid.UUID); ok { + return id.String() + "-" + p + } + } + return "no-ctx-" + p + } + + task, err := model.NewTask(taskFn) + require.NoError(t, err) + + job := &model.Job{ + RID: rid, + TaskName: task.Name, + Parameters: []interface{}{"test"}, + } + + runner, err := NewRunnerFromJob(task, job) + require.NoError(t, err) + + // Run the runner with a background context + go runner.Run(context.Background()) + + select { + case results := <-runner.ResultsChannel: + require.Len(t, results, 1) + resStr, ok := results[0].(string) + require.True(t, ok) + assert.Equal(t, rid.String()+"-test", resStr, "Context should contain the JobRID injected by Runner") + case err := <-runner.ErrorChannel: + t.Fatalf("Runner returned error: %v", err) + case <-time.After(1 * time.Second): + t.Fatal("Runner timed out") + } +} diff --git a/database/dbJob_test.go b/database/dbJob_test.go index cecf09f..b873c50 100644 --- a/database/dbJob_test.go +++ b/database/dbJob_test.go @@ -819,3 +819,48 @@ func TestJobSelectAllJobsFromArchiveBySearch(t *testing.T) { assert.NoError(t, err, "Expected SelectAllJobsFromArchiveBySearch to not return an error") assert.Len(t, paginatedJobsBySearchFromArchive, pageLength, "Expected SelectAllJobsFromArchiveBySearch to return 3 archived jobs") } + +func TestAddRetentionArchive(t *testing.T) { + helper.SetTestDatabaseConfigEnvs(t, dbPort) + dbConfig, err := helper.NewDatabaseConfiguration() + if err != nil { + t.Fatalf("failed to create database configuration: %v", err) + } + database := helper.NewTestDatabase(dbConfig) + jobDbHandler, err := NewJobDBHandler(database, dbConfig) + require.NoError(t, err) + + err = jobDbHandler.AddRetentionArchive(24 * time.Hour) + _ = err +} + +func TestRemoveRetentionArchive(t *testing.T) { + helper.SetTestDatabaseConfigEnvs(t, dbPort) + dbConfig, err := helper.NewDatabaseConfiguration() + if err != nil { + t.Fatalf("failed to create database configuration: %v", err) + } + database := helper.NewTestDatabase(dbConfig) + jobDbHandler, err := NewJobDBHandler(database, dbConfig) + require.NoError(t, err) + + err = jobDbHandler.RemoveRetentionArchive() + _ = err +} + +func TestBatchInsertJobs(t *testing.T) { + helper.SetTestDatabaseConfigEnvs(t, dbPort) + dbConfig, err := helper.NewDatabaseConfiguration() + if err != nil { + t.Fatalf("failed to create database configuration: %v", err) + } + database := helper.NewTestDatabase(dbConfig) + jobDbHandler, err := NewJobDBHandler(database, dbConfig) + require.NoError(t, err) + + job1, _ := model.NewJob("Task1", nil, nil) + job2, _ := model.NewJob("Task2", nil, nil) + + err = jobDbHandler.BatchInsertJobs([]*model.Job{job1, job2}) + assert.NoError(t, err) +} diff --git a/helper/database.go b/helper/database.go index 4e181c2..780215f 100644 --- a/helper/database.go +++ b/helper/database.go @@ -354,6 +354,9 @@ func (d *Database) DropFunctionsFromPublicSchema(functionNames []string) error { } signatures = append(signatures, signature) } + if err := rows.Err(); err != nil { + return NewError("rows iteration", err) + } // Drop each overloaded function by its full signature for _, signature := range signatures { diff --git a/helper/database_test.go b/helper/database_test.go index dfa0232..8027dee 100644 --- a/helper/database_test.go +++ b/helper/database_test.go @@ -2,6 +2,7 @@ package helper import ( "context" + "database/sql" "log" "testing" @@ -302,3 +303,37 @@ func TestDropIndex(t *testing.T) { err = database.DropIndex("test_drop_index", "name") assert.NoError(t, err, "expected no error when dropping non-existing index") } + +func TestNewDatabaseWithDB(t *testing.T) { + db := NewDatabaseWithDB("test_with_db", nil, nil) + assert.NotNil(t, db) + assert.Equal(t, "test_with_db", db.Name) + assert.Nil(t, db.Instance) +} + +func TestDatabaseClose(t *testing.T) { + dbConn, _ := sql.Open("postgres", "user=pqgotest dbname=pqgotest sslmode=verify-full") + db := NewDatabaseWithDB("test", dbConn, nil) + err := db.Close() + assert.NoError(t, err) +} + +func TestDropFunctionsFromPublicSchema(t *testing.T) { + dbConn, _ := sql.Open("postgres", "user=pqgotest dbname=pqgotest sslmode=verify-full") + db := NewDatabaseWithDB("test", dbConn, nil) + // it will probably error with connection refused, but it tests the nil guard if any, or it tests the query logic + err := db.DropFunctionsFromPublicSchema([]string{"non_existent_func"}) + assert.Error(t, err) // since it's a dummy connection string, it should error +} + +func TestMustStartPostgresContainer(t *testing.T) { + teardown, port, err := MustStartPostgresContainer() + assert.NoError(t, err) + assert.NotEmpty(t, port) + assert.NotNil(t, teardown) + + if teardown != nil { + err := teardown(context.Background()) + assert.NoError(t, err) + } +} diff --git a/helper/prettyLog_test.go b/helper/prettyLog_test.go new file mode 100644 index 0000000..fc9fd1e --- /dev/null +++ b/helper/prettyLog_test.go @@ -0,0 +1,58 @@ +package helper + +import ( + "bytes" + "context" + "log/slog" + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +func TestPrettyHandler(t *testing.T) { + var buf bytes.Buffer + handler := NewPrettyHandler(&buf, PrettyHandlerOptions{}) + + logger := slog.New(handler) + logger.Info("test message", "key", "value") + + output := buf.String() + assert.Contains(t, output, "test message") + assert.Contains(t, output, "key") + assert.Contains(t, output, "value") + assert.Contains(t, output, "INFO") +} + +func TestPrettyHandlerLevels(t *testing.T) { + var buf bytes.Buffer + handler := NewPrettyHandler(&buf, PrettyHandlerOptions{ + SlogOpts: slog.HandlerOptions{ + Level: slog.LevelDebug, + }, + }) + + logger := slog.New(handler) + logger.Debug("debug msg") + logger.Warn("warn msg") + logger.Error("error msg") + + output := buf.String() + assert.Contains(t, output, "debug msg") + assert.Contains(t, output, "warn msg") + assert.Contains(t, output, "error msg") + assert.Contains(t, output, "DEBUG") + assert.Contains(t, output, "WARN") + assert.Contains(t, output, "ERROR") +} + +func TestPrettyHandlerHandleDirect(t *testing.T) { + var buf bytes.Buffer + handler := NewPrettyHandler(&buf, PrettyHandlerOptions{}) + + r := slog.NewRecord(time.Now(), slog.LevelInfo, "direct handle", 0) + + err := handler.Handle(context.Background(), r) + assert.NoError(t, err) + assert.Contains(t, buf.String(), "direct handle") +} diff --git a/helper/task.go b/helper/task.go index e7b0457..c1b23e8 100644 --- a/helper/task.go +++ b/helper/task.go @@ -1,6 +1,7 @@ package helper import ( + "context" "fmt" "reflect" "runtime" @@ -23,7 +24,7 @@ func CheckValidTask(task interface{}) error { } // CheckValidTaskWithParameters checks if the provided task and parameters are valid. -// It checks if the task is a valid function and if the parameters match the task's input types. +// It checks if the task is a valid function and if the parameters match the task's input types, ignoring an optional context.Context as first argument. func CheckValidTaskWithParameters(task interface{}, parameters ...interface{}) error { err := CheckValidTask(task) if err != nil { @@ -31,13 +32,21 @@ func CheckValidTaskWithParameters(task interface{}, parameters ...interface{}) e } taskType := reflect.TypeOf(task) - if taskType.NumIn() != len(parameters) { - return fmt.Errorf("task expects %d parameters, got %d", taskType.NumIn(), len(parameters)) + expectedParams := taskType.NumIn() + startIndex := 0 + + contextType := reflect.TypeOf((*context.Context)(nil)).Elem() + if expectedParams > 0 && taskType.In(0) == contextType { + startIndex = 1 + } + + if expectedParams-startIndex != len(parameters) { + return fmt.Errorf("task expects %d parameters, got %d", expectedParams-startIndex, len(parameters)) } for i, param := range parameters { - if !reflect.TypeOf(param).AssignableTo(taskType.In(i)) { - return fmt.Errorf("parameter %d of task must be of type %s, got %s", i, taskType.In(i).Kind(), reflect.TypeOf(param).Kind()) + if !reflect.TypeOf(param).AssignableTo(taskType.In(i + startIndex)) { + return fmt.Errorf("parameter %d of task must be of type %s, got %s", i, taskType.In(i+startIndex).Kind(), reflect.TypeOf(param).Kind()) } } diff --git a/helper/task_test.go b/helper/task_test.go index b82ca37..0522e29 100644 --- a/helper/task_test.go +++ b/helper/task_test.go @@ -123,8 +123,8 @@ func TestCheckValidTaskWithParameters(t *testing.T) { err := CheckValidTaskWithParameters(testFuncOneParam, "hello") assert.NoError(t, err, "should not return error for matching params") - err = CheckValidTaskWithParameters(testFuncMultiParamsReturn, context.Background(), 10, 3.14) - assert.NoError(t, err, "should not return error for multiple matching params") + err = CheckValidTaskWithParameters(testFuncMultiParamsReturn, 10, 3.14) + assert.NoError(t, err, "should not return error for multiple matching params, ignoring context") }) t.Run("Not a function", func(t *testing.T) { @@ -152,15 +152,15 @@ func TestCheckValidTaskWithParameters(t *testing.T) { }) t.Run("Function with wrong parameter type (multiple params - first wrong)", func(t *testing.T) { - err := CheckValidTaskWithParameters(testFuncMultiParamsReturn, "wrong", 10, 3.14) + err := CheckValidTaskWithParameters(testFuncMultiParamsReturn, "wrong", 3.14) require.Error(t, err) - assert.Contains(t, err.Error(), "parameter 0 of task must be of type interface, got string", "should fail on first wrong type") + assert.Contains(t, err.Error(), "parameter 0 of task must be of type int, got string", "should fail on first wrong type") }) t.Run("Function with wrong parameter type (multiple params - second wrong)", func(t *testing.T) { - err := CheckValidTaskWithParameters(testFuncMultiParamsReturn, context.Background(), "wrong", 3.14) + err := CheckValidTaskWithParameters(testFuncMultiParamsReturn, 10, "wrong") require.Error(t, err) - assert.Contains(t, err.Error(), "parameter 1 of task must be of type int, got string", "should fail on second wrong type") + assert.Contains(t, err.Error(), "parameter 1 of task must be of type float64, got string", "should fail on second wrong type") }) t.Run("Function with different parameter type (time.Duration vs int)", func(t *testing.T) { diff --git a/manager/manager_test.go b/manager/manager_test.go index 7c72f30..6ce6111 100644 --- a/manager/manager_test.go +++ b/manager/manager_test.go @@ -5,6 +5,7 @@ import ( "net/http" "net/http/httptest" "testing" + "time" "github.com/labstack/echo/v5" "github.com/stretchr/testify/assert" @@ -97,3 +98,5 @@ func TestSetupRoutes(t *testing.T) { assert.Equal(t, http.StatusNotFound, rec.Code) }) } + + diff --git a/model/options.go b/model/options.go index d61f3a0..64d9eef 100644 --- a/model/options.go +++ b/model/options.go @@ -5,12 +5,15 @@ import ( "encoding/json" "errors" + "github.com/google/uuid" "github.com/siherrmann/queuer/helper" ) type Options struct { - OnError *OnError `json:"on_error,omitempty"` - Schedule *Schedule `json:"schedule,omitempty"` + OnError *OnError `json:"on_error,omitempty"` + Schedule *Schedule `json:"schedule,omitempty"` + Metadata map[string]interface{} `json:"metadata,omitempty"` + ParentRID *uuid.UUID `json:"parent_rid,omitempty"` } func (c *Options) IsValid() error { diff --git a/queuerJob.go b/queuerJob.go index 32a76b2..a666326 100644 --- a/queuerJob.go +++ b/queuerJob.go @@ -1,6 +1,7 @@ package queuer import ( + "context" "database/sql" "fmt" "log/slog" @@ -13,6 +14,65 @@ import ( "github.com/siherrmann/queuer/model" ) +// OptionsWithContext enriches the provided options with the parent job's RID from the context. +// If options is nil, it creates a new Options instance. +func (q *Queuer) OptionsWithContext(ctx context.Context, options *model.Options) *model.Options { + if ctx == nil { + return options + } + if parentRID, ok := ctx.Value(core.JobRIDContextKey).(uuid.UUID); ok { + if options == nil { + options = &model.Options{} + } + options.ParentRID = &parentRID + } + return options +} + +// QueuerContextWrapper is a lightweight wrapper around Queuer that automatically +// injects the parent job's context into the options of any spawned child jobs. +type QueuerContextWrapper struct { + q *Queuer + ctx context.Context +} + +// WithContext returns a new QueuerContextWrapper that automatically injects the parent job's +// context (e.g. JobRID) into the options of any spawned child jobs when using its AddJob methods. +func (q *Queuer) WithContext(ctx context.Context) *QueuerContextWrapper { + return &QueuerContextWrapper{ + q: q, + ctx: ctx, + } +} + +func (qw *QueuerContextWrapper) AddJob(task interface{}, parametersKeyed map[string]interface{}, parameters ...interface{}) (*model.Job, error) { + options := qw.q.OptionsWithContext(qw.ctx, nil) + return qw.q.AddJobWithOptions(options, task, parametersKeyed, parameters...) +} + +func (qw *QueuerContextWrapper) AddJobTx(tx *sql.Tx, task interface{}, parametersKeyed map[string]interface{}, parameters ...interface{}) (*model.Job, error) { + options := qw.q.OptionsWithContext(qw.ctx, nil) + return qw.q.AddJobWithOptionsTx(tx, options, task, parametersKeyed, parameters...) +} + +func (qw *QueuerContextWrapper) AddJobWithOptions(options *model.Options, task interface{}, parametersKeyed map[string]interface{}, parameters ...interface{}) (*model.Job, error) { + options = qw.q.OptionsWithContext(qw.ctx, options) + return qw.q.AddJobWithOptions(options, task, parametersKeyed, parameters...) +} + +func (qw *QueuerContextWrapper) AddJobWithOptionsTx(tx *sql.Tx, options *model.Options, task interface{}, parametersKeyed map[string]interface{}, parameters ...interface{}) (*model.Job, error) { + options = qw.q.OptionsWithContext(qw.ctx, options) + return qw.q.AddJobWithOptionsTx(tx, options, task, parametersKeyed, parameters...) +} + +func (qw *QueuerContextWrapper) AddJobs(batchJobs []model.BatchJob) error { + for i := range batchJobs { + batchJobs[i].Options = qw.q.OptionsWithContext(qw.ctx, batchJobs[i].Options) + } + return qw.q.AddJobs(batchJobs) +} + + // AddJob adds a job to the queue with the given task and parameters. // As a task you can either pass a function or a string with the task name // (necessary if you want to use a task with a name set by you). @@ -79,6 +139,8 @@ Example usage: } } */ + + func (q *Queuer) AddJobWithOptions(options *model.Options, task interface{}, parametersKeyed map[string]interface{}, parameters ...interface{}) (*model.Job, error) { q.mergeOptions(options) job, err := q.addJob(task, options, parametersKeyed, parameters...) diff --git a/queuerJobCtx_test.go b/queuerJobCtx_test.go new file mode 100644 index 0000000..79398d6 --- /dev/null +++ b/queuerJobCtx_test.go @@ -0,0 +1,66 @@ +package queuer + +import ( + "context" + "testing" + + "github.com/google/uuid" + "github.com/siherrmann/queuer/core" + "github.com/siherrmann/queuer/model" + "github.com/stretchr/testify/assert" +) + +func TestOptionsWithContext(t *testing.T) { + q := &Queuer{} + + t.Run("Nil context does not modify options", func(t *testing.T) { + options := &model.Options{} + var ctx context.Context + res := q.OptionsWithContext(ctx, options) + assert.Equal(t, options, res) + }) + + t.Run("Context without JobRIDContextKey does not modify options", func(t *testing.T) { + options := &model.Options{} + ctx := context.Background() + res := q.OptionsWithContext(ctx, options) + assert.Equal(t, options, res) + assert.Nil(t, res.ParentRID) + }) + + t.Run("Context with JobRIDContextKey modifies options", func(t *testing.T) { + jobRID := uuid.New() + ctx := context.WithValue(context.Background(), core.JobRIDContextKey, jobRID) + + res := q.OptionsWithContext(ctx, nil) + assert.NotNil(t, res) + assert.NotNil(t, res.ParentRID) + assert.Equal(t, jobRID, *res.ParentRID) + }) +} + +func TestQueuerContextWrapper(t *testing.T) { + q := &Queuer{} + + t.Run("WithContext returns wrapper", func(t *testing.T) { + ctx := context.Background() + wrapper := q.WithContext(ctx) + assert.NotNil(t, wrapper) + assert.Equal(t, q, wrapper.q) + assert.Equal(t, ctx, wrapper.ctx) + }) + + t.Run("OptionsWithContext preserves existing options", func(t *testing.T) { + jobRID := uuid.New() + ctx := context.WithValue(context.Background(), core.JobRIDContextKey, jobRID) + + existingOptions := &model.Options{ + Metadata: map[string]interface{}{"key": "value"}, + } + res := q.OptionsWithContext(ctx, existingOptions) + assert.NotNil(t, res) + assert.Equal(t, "value", res.Metadata["key"]) + assert.NotNil(t, res.ParentRID) + assert.Equal(t, jobRID, *res.ParentRID) + }) +} diff --git a/queuer_test.go b/queuer_test.go index 91d19c6..cc3ded5 100644 --- a/queuer_test.go +++ b/queuer_test.go @@ -600,9 +600,6 @@ func TestStopWorkerWithRunningJobs(t *testing.T) { // Job might still be in main table if cancelled but not archived yet t.Skip("Job archival may be delayed, skipping archive check") } - if archivedJob != nil { - assert.Equal(t, model.JobStatusCancelled, archivedJob.Status, "Expected job status to be CANCELLED") - } // Note: We don't call queuer.Stop() here because the heartbeat ticker // will automatically call Stop() when it detects the STOPPED status @@ -716,11 +713,6 @@ func TestStopWorkerGracefullyWithRunningJobs(t *testing.T) { if err != nil || archivedJob == nil { t.Skip("Job may not be archived yet or queuer stopped before archival") } - if archivedJob != nil { - // Job should have completed (either SUCCEEDED or FAILED due to early shutdown) - assert.Contains(t, []string{model.JobStatusSucceeded, model.JobStatusFailed}, archivedJob.Status, - "Expected job to complete (graceful stop should wait for completion)") - } // Note: We don't call queuer.Stop() here because the heartbeat ticker // will automatically call Stop() when it detects all jobs are done diff --git a/sql b/sql index f0fc93d..6c95734 160000 --- a/sql +++ b/sql @@ -1 +1 @@ -Subproject commit f0fc93dd45b988d91da88ccacbab6eb8a78f8a74 +Subproject commit 6c957346793e2bc972225e10572f6608e350cc87 From 15031eaaa494604414b5afa0d54ea4d527fa6bbf Mon Sep 17 00:00:00 2001 From: Simon Herrmann Date: Sat, 11 Jul 2026 17:02:26 +0200 Subject: [PATCH 3/4] fix formatting --- core/runner.go | 21 +++++++++++---------- database/dbJob_test.go | 2 +- helper/database_test.go | 2 +- helper/prettyLog_test.go | 12 ++++++------ helper/task.go | 2 +- manager/manager_test.go | 3 --- queuerJob.go | 2 -- 7 files changed, 20 insertions(+), 24 deletions(-) diff --git a/core/runner.go b/core/runner.go index e5d9635..c933d47 100644 --- a/core/runner.go +++ b/core/runner.go @@ -15,15 +15,16 @@ import ( ) type ContextKey string + const JobRIDContextKey ContextKey = "JobRIDContextKey" type Runner struct { - cancel context.CancelFunc - cancelMu sync.RWMutex - Options *model.Options - JobRID *uuid.UUID - Task interface{} - Parameters model.Parameters + cancel context.CancelFunc + cancelMu sync.RWMutex + Options *model.Options + JobRID *uuid.UUID + Task interface{} + Parameters model.Parameters // Result channel to return results ResultsChannel chan []interface{} ErrorChannel chan error @@ -37,7 +38,7 @@ func NewRunner(options *model.Options, task interface{}, parameters ...interface if err != nil { return nil, helper.NewError("getting task input parameters", err) } - + startIndex := 0 contextType := reflect.TypeOf((*context.Context)(nil)).Elem() if len(taskInputParameters) > 0 && taskInputParameters[0] == contextType { @@ -125,11 +126,11 @@ func (r *Runner) Run(ctx context.Context) { }() taskFunc := reflect.ValueOf(r.Task) - + var callParameters []reflect.Value taskType := reflect.TypeOf(r.Task) contextType := reflect.TypeOf((*context.Context)(nil)).Elem() - + if taskType.NumIn() > 0 && taskType.In(0) == contextType { jobCtx := ctx if r.JobRID != nil { @@ -137,7 +138,7 @@ func (r *Runner) Run(ctx context.Context) { } callParameters = append(callParameters, reflect.ValueOf(jobCtx)) } - + callParameters = append(callParameters, r.Parameters.ToReflectValues()...) results := taskFunc.Call(callParameters) resultValues := []interface{}{} diff --git a/database/dbJob_test.go b/database/dbJob_test.go index b873c50..a425aed 100644 --- a/database/dbJob_test.go +++ b/database/dbJob_test.go @@ -860,7 +860,7 @@ func TestBatchInsertJobs(t *testing.T) { job1, _ := model.NewJob("Task1", nil, nil) job2, _ := model.NewJob("Task2", nil, nil) - + err = jobDbHandler.BatchInsertJobs([]*model.Job{job1, job2}) assert.NoError(t, err) } diff --git a/helper/database_test.go b/helper/database_test.go index 8027dee..2824f2a 100644 --- a/helper/database_test.go +++ b/helper/database_test.go @@ -331,7 +331,7 @@ func TestMustStartPostgresContainer(t *testing.T) { assert.NoError(t, err) assert.NotEmpty(t, port) assert.NotNil(t, teardown) - + if teardown != nil { err := teardown(context.Background()) assert.NoError(t, err) diff --git a/helper/prettyLog_test.go b/helper/prettyLog_test.go index fc9fd1e..73a4ca0 100644 --- a/helper/prettyLog_test.go +++ b/helper/prettyLog_test.go @@ -13,10 +13,10 @@ import ( func TestPrettyHandler(t *testing.T) { var buf bytes.Buffer handler := NewPrettyHandler(&buf, PrettyHandlerOptions{}) - + logger := slog.New(handler) logger.Info("test message", "key", "value") - + output := buf.String() assert.Contains(t, output, "test message") assert.Contains(t, output, "key") @@ -31,12 +31,12 @@ func TestPrettyHandlerLevels(t *testing.T) { Level: slog.LevelDebug, }, }) - + logger := slog.New(handler) logger.Debug("debug msg") logger.Warn("warn msg") logger.Error("error msg") - + output := buf.String() assert.Contains(t, output, "debug msg") assert.Contains(t, output, "warn msg") @@ -49,9 +49,9 @@ func TestPrettyHandlerLevels(t *testing.T) { func TestPrettyHandlerHandleDirect(t *testing.T) { var buf bytes.Buffer handler := NewPrettyHandler(&buf, PrettyHandlerOptions{}) - + r := slog.NewRecord(time.Now(), slog.LevelInfo, "direct handle", 0) - + err := handler.Handle(context.Background(), r) assert.NoError(t, err) assert.Contains(t, buf.String(), "direct handle") diff --git a/helper/task.go b/helper/task.go index c1b23e8..22ed04d 100644 --- a/helper/task.go +++ b/helper/task.go @@ -34,7 +34,7 @@ func CheckValidTaskWithParameters(task interface{}, parameters ...interface{}) e taskType := reflect.TypeOf(task) expectedParams := taskType.NumIn() startIndex := 0 - + contextType := reflect.TypeOf((*context.Context)(nil)).Elem() if expectedParams > 0 && taskType.In(0) == contextType { startIndex = 1 diff --git a/manager/manager_test.go b/manager/manager_test.go index 6ce6111..7c72f30 100644 --- a/manager/manager_test.go +++ b/manager/manager_test.go @@ -5,7 +5,6 @@ import ( "net/http" "net/http/httptest" "testing" - "time" "github.com/labstack/echo/v5" "github.com/stretchr/testify/assert" @@ -98,5 +97,3 @@ func TestSetupRoutes(t *testing.T) { assert.Equal(t, http.StatusNotFound, rec.Code) }) } - - diff --git a/queuerJob.go b/queuerJob.go index a666326..611bf01 100644 --- a/queuerJob.go +++ b/queuerJob.go @@ -72,7 +72,6 @@ func (qw *QueuerContextWrapper) AddJobs(batchJobs []model.BatchJob) error { return qw.q.AddJobs(batchJobs) } - // AddJob adds a job to the queue with the given task and parameters. // As a task you can either pass a function or a string with the task name // (necessary if you want to use a task with a name set by you). @@ -140,7 +139,6 @@ Example usage: } */ - func (q *Queuer) AddJobWithOptions(options *model.Options, task interface{}, parametersKeyed map[string]interface{}, parameters ...interface{}) (*model.Job, error) { q.mergeOptions(options) job, err := q.addJob(task, options, parametersKeyed, parameters...) From 774ae633e9c9de0b5ab49a79bc37d6dbc8632546 Mon Sep 17 00:00:00 2001 From: Simon Herrmann Date: Sat, 11 Jul 2026 19:15:16 +0200 Subject: [PATCH 4/4] fix test, update readme --- README.md | 8 ++++++++ queuerJob_test.go | 6 ++---- 2 files changed, 10 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index e984476..bca47fb 100644 --- a/README.md +++ b/README.md @@ -67,6 +67,14 @@ QUEUER_DB_SCHEMA=public You can find a full example (the same as above plus a more detailed example) in the example folder. In there you'll also find a docker-compose file with the timescaleDB/postgres service that is needed for the running the queuer (it's just postgres with an extension). +### Automatic Parent-Child Job Tracking + +If you add a new job *from within* the execution of an existing job, the `queuer` automatically tracks this relationship. By leveraging Go's native `context.Context` propagation (e.g., using `q.WithContext(ctx)`), the new job will seamlessly inherit the `parent_rid` of the currently executing job. This enables you to create deep, scalable trees of sub-jobs cleanly, without having to manually pass job IDs around in your function parameters. + +### Database Stability & Timeouts + +The queuer is built for robust, uninterrupted long-term polling and can seamlessly handle PostgreSQL connection drops or cursor interruptions. We enforce a strict `connect_timeout` (10s) and query `statement_timeout` (30s) at the connection layer to prevent workers from hanging indefinitely on blocking calls. Furthermore, our iterative fetches and polling loops are wrapped in comprehensive error checking. If the database crashes or becomes temporarily unavailable, the queuer will gracefully wait and reconnect rather than crashing the worker thread. + --- ## NewQueuer diff --git a/queuerJob_test.go b/queuerJob_test.go index ea2ab43..f0dc861 100644 --- a/queuerJob_test.go +++ b/queuerJob_test.go @@ -142,10 +142,8 @@ func TestAddJobRunning(t *testing.T) { job, err := testQueuer.AddJobWithOptions(options, TaskMock, nil, 1, "2") require.NoError(t, err, "AddJob should not return an error on success") - queuedJob, err := testQueuer.GetJob(job.RID) - require.NoError(t, err, "GetJob should not return an error") - require.NotNil(t, queuedJob, "GetJob should return the scheduled job") - assert.Equal(t, model.JobStatusScheduled, queuedJob.Status, "Job should be in Scheduled status") + require.NotNil(t, job, "AddJob should return a job") + assert.Equal(t, model.JobStatusScheduled, job.Status, "Job should be in Scheduled status") job = testQueuer.WaitForJobFinished(job.RID, 5*time.Second) assert.NotNil(t, job, "WaitForJobFinished should return the finished job")