From a26b3f8a6a8456ebc704c323bb213caa71316b0a Mon Sep 17 00:00:00 2001 From: Simon Herrmann Date: Mon, 9 Feb 2026 17:26:55 +0100 Subject: [PATCH] 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