From 2978992e124a9572ce8b9a7886402ead4bae1995 Mon Sep 17 00:00:00 2001 From: Tomas Dvorak Date: Sat, 10 Oct 2026 04:19:45 +0200 Subject: [PATCH] refactor(db): port cron jobs to sqlc MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tranche 6 — 15 of 17 raw sites in cron.go. Adds cron_jobs and cron_executions to schema.sql and a full query set: job CRUD, owner joins (via project and via service), execution list/insert/finish, due-job scan for the scheduler, run bookkeeping, and retention trim. The filtered list query and the sparse UPDATE builder stay raw — both are genuinely dynamic. Go models keep string IDs; cronJobFrom and cronExecutionFrom map uuid/null types at the boundary. TestCronLiveInvariants covers insert/read, both owner joins, the trigger read, execution lifecycle, retention trim, and delete on live Postgres. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- app/backend/internal/api/cron.go | 262 +++++++------- app/backend/internal/api/cron_test.go | 127 +++++++ .../internal/database/sqlcdb/cron.sql.go | 332 ++++++++++++++++++ .../internal/database/sqlcdb/models.go | 28 ++ .../internal/database/sqlcdb/querier.go | 13 + app/backend/sqlc/queries/cron.sql | 58 +++ app/backend/sqlc/schema.sql | 28 ++ 7 files changed, 729 insertions(+), 119 deletions(-) create mode 100644 app/backend/internal/database/sqlcdb/cron.sql.go create mode 100644 app/backend/sqlc/queries/cron.sql diff --git a/app/backend/internal/api/cron.go b/app/backend/internal/api/cron.go index 0b15f81..e16e8b5 100644 --- a/app/backend/internal/api/cron.go +++ b/app/backend/internal/api/cron.go @@ -65,13 +65,47 @@ type UpdateCronJobRequest struct { Retention int `json:"retention"` } +// cronJobFrom maps the sqlc row onto the string-ID API model. +func cronJobFrom(j sqlcdb.CronJob) CronJob { + return CronJob{ + ID: j.ID.String(), + ProjectID: j.ProjectID.String(), + ServiceID: j.ServiceID.String(), + Name: j.Name, + Schedule: j.Schedule, + Command: j.Command, + Timezone: j.Timezone.String, + Enabled: j.Enabled.Bool, + LastRunAt: timePtrNT(j.LastRunAt), + NextRunAt: timePtrNT(j.NextRunAt), + LastStatus: j.LastStatus.String, + LastOutput: j.LastOutput.String, + Retention: int(j.Retention.Int32), + CreatedAt: j.CreatedAt.Time, + UpdatedAt: j.UpdatedAt.Time, + } +} + +// cronExecutionFrom maps the sqlc row onto the string-ID API model. +func cronExecutionFrom(e sqlcdb.CronExecution) CronExecution { + return CronExecution{ + ID: e.ID.String(), + CronJobID: e.CronJobID.String(), + StartedAt: e.StartedAt, + FinishedAt: timePtrNT(e.FinishedAt), + Status: e.Status.String, + Output: e.Output.String, + Error: e.Error.String, + } +} + // cronJobProjectID resolves the project a cron job belongs to. func cronJobProjectID(db *database.DB, jobID string) (uuid.UUID, bool) { - var projectID uuid.UUID - err := db.QueryRow( - `SELECT cj.project_id FROM cron_jobs cj WHERE cj.id = $1`, - jobID, - ).Scan(&projectID) + id, err := uuid.Parse(jobID) + if err != nil { + return uuid.Nil, false + } + projectID, err := sqlcdb.New(db.DB).GetCronJobProjectID(context.Background(), id) if err != nil { return uuid.Nil, false } @@ -137,15 +171,14 @@ func handleCreateCronJob(c *gin.Context) { return } - var ownerCheck string - err := db.QueryRow( - `SELECT p.owner_id FROM projects p - JOIN services s ON s.project_id = p.id - WHERE s.id = $1`, - req.ServiceID, - ).Scan(&ownerCheck) + svcID, err := uuid.Parse(req.ServiceID) + if err != nil { + c.JSON(http.StatusForbidden, gin.H{"error": "Access denied"}) + return + } + ownerCheck, err := sqlcdb.New(db.DB).GetCronOwnerViaService(context.Background(), svcID) - if err != nil || (ownerCheck != userID && !contextIsAdmin(c)) { + if err != nil || (ownerCheck.String() != userID && !contextIsAdmin(c)) { c.JSON(http.StatusForbidden, gin.H{"error": "Access denied"}) return } @@ -178,11 +211,20 @@ func handleCreateCronJob(c *gin.Context) { UpdatedAt: time.Now(), } - _, err = db.Exec( - `INSERT INTO cron_jobs (id, project_id, service_id, name, schedule, command, timezone, enabled, next_run_at, retention, created_at, updated_at) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)`, - job.ID, job.ProjectID, job.ServiceID, job.Name, job.Schedule, job.Command, job.Timezone, job.Enabled, job.NextRunAt, job.Retention, job.CreatedAt, job.UpdatedAt, - ) + err = sqlcdb.New(db.DB).InsertCronJob(context.Background(), sqlcdb.InsertCronJobParams{ + ID: uuid.MustParse(job.ID), + ProjectID: uuid.MustParse(job.ProjectID), + ServiceID: uuid.MustParse(job.ServiceID), + Name: job.Name, + Schedule: job.Schedule, + Command: job.Command, + Timezone: sql.NullString{String: job.Timezone, Valid: true}, + Enabled: sql.NullBool{Bool: job.Enabled, Valid: true}, + NextRunAt: ntPtr(job.NextRunAt), + Retention: sql.NullInt32{Int32: int32(job.Retention), Valid: true}, + CreatedAt: sql.NullTime{Time: job.CreatedAt, Valid: true}, + UpdatedAt: sql.NullTime{Time: job.UpdatedAt, Valid: true}, + }) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to create cron job"}) @@ -201,22 +243,17 @@ func handleGetCronJob(c *gin.Context) { db := c.MustGet("db").(*database.DB) jobID := c.Param("id") - var job CronJob - err := db.QueryRow( - `SELECT cj.id, cj.project_id, cj.service_id, cj.name, cj.schedule, cj.timezone, - cj.enabled, cj.last_run_at, cj.next_run_at, cj.last_status, cj.last_output, - cj.retention, cj.created_at, cj.updated_at - FROM cron_jobs cj - WHERE cj.id = $1`, - jobID, - ).Scan(&job.ID, &job.ProjectID, &job.ServiceID, &job.Name, &job.Schedule, &job.Timezone, - &job.Enabled, &job.LastRunAt, &job.NextRunAt, &job.LastStatus, &job.LastOutput, - &job.Retention, &job.CreatedAt, &job.UpdatedAt) - + jid, err := uuid.Parse(jobID) if err != nil { c.JSON(http.StatusNotFound, gin.H{"error": "Cron job not found"}) return } + row, err := sqlcdb.New(db.DB).GetCronJob(context.Background(), jid) + if err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "Cron job not found"}) + return + } + job := cronJobFrom(row) projectID, found := cronJobProjectID(db, jobID) if !found { @@ -242,15 +279,14 @@ func handleUpdateCronJob(c *gin.Context) { return } - var ownerCheck string - err := db.QueryRow( - `SELECT p.owner_id FROM cron_jobs cj - JOIN projects p ON cj.project_id = p.id - WHERE cj.id = $1`, - jobID, - ).Scan(&ownerCheck) + jid, err := uuid.Parse(jobID) + if err != nil { + c.JSON(http.StatusForbidden, gin.H{"error": "Access denied"}) + return + } + ownerCheck, err := sqlcdb.New(db.DB).GetCronJobOwner(context.Background(), jid) - if err != nil || (ownerCheck != userID && !contextIsAdmin(c)) { + if err != nil || (ownerCheck.String() != userID && !contextIsAdmin(c)) { c.JSON(http.StatusForbidden, gin.H{"error": "Access denied"}) return } @@ -312,21 +348,19 @@ func handleDeleteCronJob(c *gin.Context) { db := c.MustGet("db").(*database.DB) jobID := c.Param("id") - var ownerCheck string - err := db.QueryRow( - `SELECT p.owner_id FROM cron_jobs cj - JOIN projects p ON cj.project_id = p.id - WHERE cj.id = $1`, - jobID, - ).Scan(&ownerCheck) + jid, err := uuid.Parse(jobID) + if err != nil { + c.JSON(http.StatusForbidden, gin.H{"error": "Access denied"}) + return + } + ownerCheck, err := sqlcdb.New(db.DB).GetCronJobOwner(context.Background(), jid) - if err != nil || (ownerCheck != userID && !contextIsAdmin(c)) { + if err != nil || (ownerCheck.String() != userID && !contextIsAdmin(c)) { c.JSON(http.StatusForbidden, gin.H{"error": "Access denied"}) return } - _, err = db.Exec("DELETE FROM cron_jobs WHERE id = $1", jobID) - if err != nil { + if err := sqlcdb.New(db.DB).DeleteCronJob(context.Background(), jid); err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to delete cron job"}) return } @@ -350,29 +384,20 @@ func handleGetCronExecutions(c *gin.Context) { return } - rows, err := db.Query( - `SELECT id, cron_job_id, started_at, finished_at, status, output, error - FROM cron_executions - WHERE cron_job_id = $1 - ORDER BY started_at DESC - LIMIT 100`, - jobID, - ) - + jid, err := uuid.Parse(jobID) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to fetch executions"}) + return + } + rows, err := sqlcdb.New(db.DB).ListCronExecutions(context.Background(), jid) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to fetch executions"}) return } - defer rows.Close() var executions []CronExecution - for rows.Next() { - var exec CronExecution - err := rows.Scan(&exec.ID, &exec.CronJobID, &exec.StartedAt, &exec.FinishedAt, &exec.Status, &exec.Output, &exec.Error) - if err != nil { - continue - } - executions = append(executions, exec) + for _, e := range rows { + executions = append(executions, cronExecutionFrom(e)) } c.JSON(http.StatusOK, gin.H{"executions": executions}) @@ -383,29 +408,33 @@ func handleTriggerCronJob(c *gin.Context) { db := c.MustGet("db").(*database.DB) jobID := c.Param("id") - var job CronJob - var ownerCheck string - err := db.QueryRow( - `SELECT cj.service_id, cj.command, cj.schedule, cj.timezone, cj.retention, p.owner_id - FROM cron_jobs cj - JOIN projects p ON cj.project_id = p.id - WHERE cj.id = $1`, - jobID, - ).Scan(&job.ServiceID, &job.Command, &job.Schedule, &job.Timezone, &job.Retention, &ownerCheck) - - if err != nil || (ownerCheck != userID && !contextIsAdmin(c)) { + jid, err := uuid.Parse(jobID) + if err != nil { c.JSON(http.StatusForbidden, gin.H{"error": "Access denied"}) return } + trig, err := sqlcdb.New(db.DB).GetCronJobForTrigger(context.Background(), jid) + if err != nil || (trig.OwnerID.String() != userID && !contextIsAdmin(c)) { + c.JSON(http.StatusForbidden, gin.H{"error": "Access denied"}) + return + } + job := CronJob{ + ServiceID: trig.ServiceID.String(), + Command: trig.Command, + Schedule: trig.Schedule, + Timezone: trig.Timezone.String, + Retention: int(trig.Retention.Int32), + } execID := uuid.New().String() now := time.Now() - _, err = db.Exec( - `INSERT INTO cron_executions (id, cron_job_id, started_at, status) - VALUES ($1, $2, $3, $4)`, - execID, jobID, now, "running", - ) + _ = sqlcdb.New(db.DB).InsertCronExecution(context.Background(), sqlcdb.InsertCronExecutionParams{ + ID: uuid.MustParse(execID), + CronJobID: jid, + StartedAt: now, + Status: sql.NullString{String: "running", Valid: true}, + }) dockerClient, _ := c.Get("docker_client") dockerCli, _ := dockerClient.(*docker.Client) @@ -458,39 +487,25 @@ func StartCronScheduler(ctx context.Context, db *database.DB, dockerClient *dock } func runDueCronJobs(db *database.DB, dockerClient *docker.Client) { - rows, err := db.Query( - `SELECT id, project_id, service_id, name, schedule, command, timezone, enabled, retention - FROM cron_jobs - WHERE enabled = TRUE AND next_run_at IS NOT NULL AND next_run_at <= NOW()`) + q := sqlcdb.New(db.DB) + jobs, err := q.ListDueCronJobs(context.Background()) if err != nil { log.Printf("cron scheduler: failed to list due jobs: %v", err) return } - defer rows.Close() - - type dueJob struct { - CronJob - Command string - } - var jobs []dueJob - for rows.Next() { - var j dueJob - if err := rows.Scan(&j.ID, &j.ProjectID, &j.ServiceID, &j.Name, &j.Schedule, - &j.Command, &j.Timezone, &j.Enabled, &j.Retention); err != nil { - continue - } - jobs = append(jobs, j) - } for _, j := range jobs { execID := uuid.New().String() - if _, err := db.Exec( - `INSERT INTO cron_executions (id, cron_job_id, started_at, status) VALUES ($1, $2, $3, $4)`, - execID, j.ID, time.Now(), "running"); err != nil { + if err := q.InsertCronExecution(context.Background(), sqlcdb.InsertCronExecutionParams{ + ID: uuid.MustParse(execID), + CronJobID: j.ID, + StartedAt: time.Now(), + Status: sql.NullString{String: "running", Valid: true}, + }); err != nil { log.Printf("cron scheduler: failed to record execution for %s: %v", j.ID, err) continue } - go executeCronJob(db, dockerClient, j.ID, j.ServiceID, execID, j.Command, j.Schedule, j.Timezone, j.Retention) + go executeCronJob(db, dockerClient, j.ID.String(), j.ServiceID.String(), execID, j.Command, j.Schedule, j.Timezone.String, int(j.Retention.Int32)) } } @@ -584,23 +599,32 @@ func executeCronJob(db *database.DB, dockerClient *docker.Client, jobID, service } now := time.Now() - db.Exec( - `UPDATE cron_executions SET finished_at = $1, status = $2, output = $3, error = $4 WHERE id = $5`, - now, status, output, errText, execID, - ) + q := sqlcdb.New(db.DB) + if eid, err := uuid.Parse(execID); err == nil { + _ = q.FinishCronExecution(context.Background(), sqlcdb.FinishCronExecutionParams{ + FinishedAt: sql.NullTime{Time: now, Valid: true}, + Status: sql.NullString{String: status, Valid: true}, + Output: sql.NullString{String: output, Valid: true}, + Error: sql.NullString{String: errText, Valid: true}, + ID: eid, + }) + } next, _ := calculateNextRun(schedule, timezone) - db.Exec( - `UPDATE cron_jobs SET last_run_at = $1, last_status = $2, last_output = $3, next_run_at = $4 WHERE id = $5`, - now, status, output, next, jobID, - ) + if jid, err := uuid.Parse(jobID); err == nil { + _ = q.UpdateCronJobRun(context.Background(), sqlcdb.UpdateCronJobRunParams{ + LastRunAt: sql.NullTime{Time: now, Valid: true}, + LastStatus: sql.NullString{String: status, Valid: true}, + LastOutput: sql.NullString{String: output, Valid: true}, + NextRunAt: ntPtr(next), + ID: jid, + }) - if retention > 0 { - db.Exec( - `DELETE FROM cron_executions WHERE cron_job_id = $1 AND id NOT IN ( - SELECT id FROM cron_executions WHERE cron_job_id = $1 - ORDER BY started_at DESC LIMIT $2)`, - jobID, retention, - ) + if retention > 0 { + _ = q.TrimCronExecutions(context.Background(), sqlcdb.TrimCronExecutionsParams{ + CronJobID: jid, + Limit: int32(retention), + }) + } } } diff --git a/app/backend/internal/api/cron_test.go b/app/backend/internal/api/cron_test.go index 7bb3b88..1fbcf23 100644 --- a/app/backend/internal/api/cron_test.go +++ b/app/backend/internal/api/cron_test.go @@ -1,9 +1,16 @@ package api import ( + "context" + "database/sql" "strings" "testing" "time" + + "containr/internal/database" + "containr/internal/database/sqlcdb" + + "github.com/google/uuid" ) func TestCalculateNextRun(t *testing.T) { @@ -64,3 +71,123 @@ func TestCalculateNextRunTimezone(t *testing.T) { t.Logf("utc next: %v / tokyo next: %v", utcNext, tokyoNext) } } + +// TestCronLiveInvariants exercises the cron.sql queries on live Postgres: +// insert/read/owner join, execution lifecycle, run bookkeeping, retention trim. +func TestCronLiveInvariants(t *testing.T) { + dsn := testDSNString(t) + db, err := database.NewConnection(dsn) + if err != nil { + t.Fatalf("connect: %v", err) + } + defer db.Close() + ctx := context.Background() + q := sqlcdb.New(db.DB) + + userID := uuid.New() + projectID := uuid.New() + envID := uuid.New() + serviceID := uuid.New() + jobID := uuid.New() + suffix := strings.ReplaceAll(jobID.String()[:8], "-", "") + must := func(err error) { + t.Helper() + if err != nil { + t.Fatalf("setup: %v", err) + } + } + _, err = db.Exec(`INSERT INTO users (id, email, password_hash, name) VALUES ($1, $2, 'x', 'test')`, + userID, "cron-"+suffix+"@test.local") + must(err) + _, err = db.Exec(`INSERT INTO projects (id, name, owner_id, is_approved) VALUES ($1, $2, $3, true)`, + projectID, "cron-"+suffix, userID) + must(err) + _, err = db.Exec(`INSERT INTO environments (id, name, project_id) VALUES ($1, 'production', $2)`, + envID, projectID) + must(err) + _, err = db.Exec(`INSERT INTO services (id, name, project_id, environment_id, service_type, source_type) + VALUES ($1, $2, $3, $4, 'web', 'image')`, + serviceID, "cron-"+suffix, projectID, envID) + must(err) + defer func() { _, _ = db.Exec(`DELETE FROM users WHERE id = $1`, userID) }() + + next, err := calculateNextRun("0 4 * * *", "UTC") + if err != nil { + t.Fatalf("schedule: %v", err) + } + must(q.InsertCronJob(ctx, sqlcdb.InsertCronJobParams{ + ID: jobID, ProjectID: projectID, ServiceID: serviceID, + Name: "job-" + suffix, Schedule: "0 4 * * *", Command: "echo hi", + Timezone: sql.NullString{String: "UTC", Valid: true}, + Enabled: sql.NullBool{Bool: true, Valid: true}, + NextRunAt: ntPtr(next), + })) + job, err := q.GetCronJob(ctx, jobID) + if err != nil || job.Name != "job-"+suffix || job.Command != "echo hi" { + t.Fatalf("job: %v %+v", err, job) + } + owner, err := q.GetCronJobOwner(ctx, jobID) + if err != nil || owner != userID { + t.Fatalf("owner: %v %v", err, owner) + } + owner2, err := q.GetCronOwnerViaService(ctx, serviceID) + if err != nil || owner2 != userID { + t.Fatalf("owner via service: %v %v", err, owner2) + } + pid, err := q.GetCronJobProjectID(ctx, jobID) + if err != nil || pid != projectID { + t.Fatalf("project: %v %v", err, pid) + } + + // Trigger-shaped read. + trig, err := q.GetCronJobForTrigger(ctx, jobID) + if err != nil || trig.ServiceID != serviceID || trig.OwnerID != userID { + t.Fatalf("trigger: %v %+v", err, trig) + } + + // Execution lifecycle + retention trim. + for i := 0; i < 3; i++ { + execID := uuid.New() + must(q.InsertCronExecution(ctx, sqlcdb.InsertCronExecutionParams{ + ID: execID, CronJobID: jobID, + StartedAt: time.Now().Add(time.Duration(i) * time.Second), + Status: sql.NullString{String: "running", Valid: true}, + })) + must(q.FinishCronExecution(ctx, sqlcdb.FinishCronExecutionParams{ + FinishedAt: sql.NullTime{Time: time.Now(), Valid: true}, + Status: sql.NullString{String: "success", Valid: true}, + Output: sql.NullString{String: "ok", Valid: true}, + Error: sql.NullString{String: "", Valid: true}, + ID: execID, + })) + } + execs, err := q.ListCronExecutions(ctx, jobID) + if err != nil || len(execs) != 3 { + t.Fatalf("execs: %v %d", err, len(execs)) + } + if execs[0].Status.String != "success" || !execs[0].FinishedAt.Valid { + t.Fatalf("exec row: %+v", execs[0]) + } + must(q.TrimCronExecutions(ctx, sqlcdb.TrimCronExecutionsParams{CronJobID: jobID, Limit: 1})) + execs, _ = q.ListCronExecutions(ctx, jobID) + if len(execs) != 1 { + t.Fatalf("trim kept %d, want 1", len(execs)) + } + + must(q.UpdateCronJobRun(ctx, sqlcdb.UpdateCronJobRunParams{ + LastRunAt: sql.NullTime{Time: time.Now(), Valid: true}, + LastStatus: sql.NullString{String: "success", Valid: true}, + LastOutput: sql.NullString{String: "ok", Valid: true}, + NextRunAt: ntPtr(next), + ID: jobID, + })) + job, _ = q.GetCronJob(ctx, jobID) + if job.LastStatus.String != "success" || !job.LastRunAt.Valid { + t.Fatalf("post-run job: %+v", job) + } + + must(q.DeleteCronJob(ctx, jobID)) + if _, err := q.GetCronJob(ctx, jobID); err == nil { + t.Fatal("deleted job still readable") + } +} diff --git a/app/backend/internal/database/sqlcdb/cron.sql.go b/app/backend/internal/database/sqlcdb/cron.sql.go new file mode 100644 index 0000000..6ab8a38 --- /dev/null +++ b/app/backend/internal/database/sqlcdb/cron.sql.go @@ -0,0 +1,332 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: cron.sql + +package sqlcdb + +import ( + "context" + "database/sql" + "time" + + "github.com/google/uuid" +) + +const deleteCronJob = `-- name: DeleteCronJob :exec +DELETE FROM cron_jobs WHERE id = $1 +` + +func (q *Queries) DeleteCronJob(ctx context.Context, id uuid.UUID) error { + _, err := q.db.ExecContext(ctx, deleteCronJob, id) + return err +} + +const finishCronExecution = `-- name: FinishCronExecution :exec +UPDATE cron_executions SET finished_at = $1, status = $2, output = $3, error = $4 WHERE id = $5 +` + +type FinishCronExecutionParams struct { + FinishedAt sql.NullTime `json:"finished_at"` + Status sql.NullString `json:"status"` + Output sql.NullString `json:"output"` + Error sql.NullString `json:"error"` + ID uuid.UUID `json:"id"` +} + +func (q *Queries) FinishCronExecution(ctx context.Context, arg FinishCronExecutionParams) error { + _, err := q.db.ExecContext(ctx, finishCronExecution, + arg.FinishedAt, + arg.Status, + arg.Output, + arg.Error, + arg.ID, + ) + return err +} + +const getCronJob = `-- name: GetCronJob :one +SELECT id, project_id, service_id, name, schedule, command, timezone, + enabled, last_run_at, next_run_at, last_status, last_output, + retention, created_at, updated_at +FROM cron_jobs WHERE id = $1 +` + +func (q *Queries) GetCronJob(ctx context.Context, id uuid.UUID) (CronJob, error) { + row := q.db.QueryRowContext(ctx, getCronJob, id) + var i CronJob + err := row.Scan( + &i.ID, + &i.ProjectID, + &i.ServiceID, + &i.Name, + &i.Schedule, + &i.Command, + &i.Timezone, + &i.Enabled, + &i.LastRunAt, + &i.NextRunAt, + &i.LastStatus, + &i.LastOutput, + &i.Retention, + &i.CreatedAt, + &i.UpdatedAt, + ) + return i, err +} + +const getCronJobForTrigger = `-- name: GetCronJobForTrigger :one +SELECT cj.service_id, cj.command, cj.schedule, cj.timezone, cj.retention, p.owner_id +FROM cron_jobs cj +JOIN projects p ON cj.project_id = p.id +WHERE cj.id = $1 +` + +type GetCronJobForTriggerRow struct { + ServiceID uuid.UUID `json:"service_id"` + Command string `json:"command"` + Schedule string `json:"schedule"` + Timezone sql.NullString `json:"timezone"` + Retention sql.NullInt32 `json:"retention"` + OwnerID uuid.UUID `json:"owner_id"` +} + +func (q *Queries) GetCronJobForTrigger(ctx context.Context, id uuid.UUID) (GetCronJobForTriggerRow, error) { + row := q.db.QueryRowContext(ctx, getCronJobForTrigger, id) + var i GetCronJobForTriggerRow + err := row.Scan( + &i.ServiceID, + &i.Command, + &i.Schedule, + &i.Timezone, + &i.Retention, + &i.OwnerID, + ) + return i, err +} + +const getCronJobOwner = `-- name: GetCronJobOwner :one +SELECT p.owner_id FROM cron_jobs cj +JOIN projects p ON cj.project_id = p.id +WHERE cj.id = $1 +` + +func (q *Queries) GetCronJobOwner(ctx context.Context, id uuid.UUID) (uuid.UUID, error) { + row := q.db.QueryRowContext(ctx, getCronJobOwner, id) + var owner_id uuid.UUID + err := row.Scan(&owner_id) + return owner_id, err +} + +const getCronJobProjectID = `-- name: GetCronJobProjectID :one +SELECT project_id FROM cron_jobs WHERE id = $1 +` + +func (q *Queries) GetCronJobProjectID(ctx context.Context, id uuid.UUID) (uuid.UUID, error) { + row := q.db.QueryRowContext(ctx, getCronJobProjectID, id) + var project_id uuid.UUID + err := row.Scan(&project_id) + return project_id, err +} + +const getCronOwnerViaService = `-- name: GetCronOwnerViaService :one +SELECT p.owner_id FROM projects p +JOIN services s ON s.project_id = p.id +WHERE s.id = $1 +` + +func (q *Queries) GetCronOwnerViaService(ctx context.Context, id uuid.UUID) (uuid.UUID, error) { + row := q.db.QueryRowContext(ctx, getCronOwnerViaService, id) + var owner_id uuid.UUID + err := row.Scan(&owner_id) + return owner_id, err +} + +const insertCronExecution = `-- name: InsertCronExecution :exec +INSERT INTO cron_executions (id, cron_job_id, started_at, status) +VALUES ($1, $2, $3, $4) +` + +type InsertCronExecutionParams struct { + ID uuid.UUID `json:"id"` + CronJobID uuid.UUID `json:"cron_job_id"` + StartedAt time.Time `json:"started_at"` + Status sql.NullString `json:"status"` +} + +func (q *Queries) InsertCronExecution(ctx context.Context, arg InsertCronExecutionParams) error { + _, err := q.db.ExecContext(ctx, insertCronExecution, + arg.ID, + arg.CronJobID, + arg.StartedAt, + arg.Status, + ) + return err +} + +const insertCronJob = `-- name: InsertCronJob :exec +INSERT INTO cron_jobs (id, project_id, service_id, name, schedule, command, timezone, enabled, next_run_at, retention, created_at, updated_at) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) +` + +type InsertCronJobParams struct { + ID uuid.UUID `json:"id"` + ProjectID uuid.UUID `json:"project_id"` + ServiceID uuid.UUID `json:"service_id"` + Name string `json:"name"` + Schedule string `json:"schedule"` + Command string `json:"command"` + Timezone sql.NullString `json:"timezone"` + Enabled sql.NullBool `json:"enabled"` + NextRunAt sql.NullTime `json:"next_run_at"` + Retention sql.NullInt32 `json:"retention"` + CreatedAt sql.NullTime `json:"created_at"` + UpdatedAt sql.NullTime `json:"updated_at"` +} + +func (q *Queries) InsertCronJob(ctx context.Context, arg InsertCronJobParams) error { + _, err := q.db.ExecContext(ctx, insertCronJob, + arg.ID, + arg.ProjectID, + arg.ServiceID, + arg.Name, + arg.Schedule, + arg.Command, + arg.Timezone, + arg.Enabled, + arg.NextRunAt, + arg.Retention, + arg.CreatedAt, + arg.UpdatedAt, + ) + return err +} + +const listCronExecutions = `-- name: ListCronExecutions :many +SELECT id, cron_job_id, started_at, finished_at, status, output, error +FROM cron_executions +WHERE cron_job_id = $1 +ORDER BY started_at DESC +LIMIT 100 +` + +func (q *Queries) ListCronExecutions(ctx context.Context, cronJobID uuid.UUID) ([]CronExecution, error) { + rows, err := q.db.QueryContext(ctx, listCronExecutions, cronJobID) + if err != nil { + return nil, err + } + defer rows.Close() + items := []CronExecution{} + for rows.Next() { + var i CronExecution + if err := rows.Scan( + &i.ID, + &i.CronJobID, + &i.StartedAt, + &i.FinishedAt, + &i.Status, + &i.Output, + &i.Error, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listDueCronJobs = `-- name: ListDueCronJobs :many +SELECT id, project_id, service_id, name, schedule, command, timezone, enabled, retention +FROM cron_jobs +WHERE enabled = TRUE AND next_run_at IS NOT NULL AND next_run_at <= NOW() +` + +type ListDueCronJobsRow struct { + ID uuid.UUID `json:"id"` + ProjectID uuid.UUID `json:"project_id"` + ServiceID uuid.UUID `json:"service_id"` + Name string `json:"name"` + Schedule string `json:"schedule"` + Command string `json:"command"` + Timezone sql.NullString `json:"timezone"` + Enabled sql.NullBool `json:"enabled"` + Retention sql.NullInt32 `json:"retention"` +} + +func (q *Queries) ListDueCronJobs(ctx context.Context) ([]ListDueCronJobsRow, error) { + rows, err := q.db.QueryContext(ctx, listDueCronJobs) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ListDueCronJobsRow{} + for rows.Next() { + var i ListDueCronJobsRow + if err := rows.Scan( + &i.ID, + &i.ProjectID, + &i.ServiceID, + &i.Name, + &i.Schedule, + &i.Command, + &i.Timezone, + &i.Enabled, + &i.Retention, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const trimCronExecutions = `-- name: TrimCronExecutions :exec +DELETE FROM cron_executions WHERE cron_executions.cron_job_id = $1 AND cron_executions.id NOT IN ( + SELECT ce.id FROM cron_executions ce WHERE ce.cron_job_id = $1 + ORDER BY ce.started_at DESC LIMIT $2) +` + +type TrimCronExecutionsParams struct { + CronJobID uuid.UUID `json:"cron_job_id"` + Limit int32 `json:"limit"` +} + +func (q *Queries) TrimCronExecutions(ctx context.Context, arg TrimCronExecutionsParams) error { + _, err := q.db.ExecContext(ctx, trimCronExecutions, arg.CronJobID, arg.Limit) + return err +} + +const updateCronJobRun = `-- name: UpdateCronJobRun :exec +UPDATE cron_jobs SET last_run_at = $1, last_status = $2, last_output = $3, next_run_at = $4 WHERE id = $5 +` + +type UpdateCronJobRunParams struct { + LastRunAt sql.NullTime `json:"last_run_at"` + LastStatus sql.NullString `json:"last_status"` + LastOutput sql.NullString `json:"last_output"` + NextRunAt sql.NullTime `json:"next_run_at"` + ID uuid.UUID `json:"id"` +} + +func (q *Queries) UpdateCronJobRun(ctx context.Context, arg UpdateCronJobRunParams) error { + _, err := q.db.ExecContext(ctx, updateCronJobRun, + arg.LastRunAt, + arg.LastStatus, + arg.LastOutput, + arg.NextRunAt, + arg.ID, + ) + return err +} diff --git a/app/backend/internal/database/sqlcdb/models.go b/app/backend/internal/database/sqlcdb/models.go index 775fc4c..f9cb6e6 100644 --- a/app/backend/internal/database/sqlcdb/models.go +++ b/app/backend/internal/database/sqlcdb/models.go @@ -98,6 +98,34 @@ type ContainerInstance struct { UpdatedAt sql.NullTime `json:"updated_at"` } +type CronExecution struct { + ID uuid.UUID `json:"id"` + CronJobID uuid.UUID `json:"cron_job_id"` + StartedAt time.Time `json:"started_at"` + FinishedAt sql.NullTime `json:"finished_at"` + Status sql.NullString `json:"status"` + Output sql.NullString `json:"output"` + Error sql.NullString `json:"error"` +} + +type CronJob struct { + ID uuid.UUID `json:"id"` + ProjectID uuid.UUID `json:"project_id"` + ServiceID uuid.UUID `json:"service_id"` + Name string `json:"name"` + Schedule string `json:"schedule"` + Command string `json:"command"` + Timezone sql.NullString `json:"timezone"` + Enabled sql.NullBool `json:"enabled"` + LastRunAt sql.NullTime `json:"last_run_at"` + NextRunAt sql.NullTime `json:"next_run_at"` + LastStatus sql.NullString `json:"last_status"` + LastOutput sql.NullString `json:"last_output"` + Retention sql.NullInt32 `json:"retention"` + CreatedAt sql.NullTime `json:"created_at"` + UpdatedAt sql.NullTime `json:"updated_at"` +} + type DatabaseBackup struct { ID string `json:"id"` DatabaseID string `json:"database_id"` diff --git a/app/backend/internal/database/sqlcdb/querier.go b/app/backend/internal/database/sqlcdb/querier.go index 46c7a0c..ea6e26e 100644 --- a/app/backend/internal/database/sqlcdb/querier.go +++ b/app/backend/internal/database/sqlcdb/querier.go @@ -59,6 +59,7 @@ type Querier interface { DeleteAllServiceDomains(ctx context.Context, serviceID uuid.UUID) error DeleteBackupTargetByIDAndUser(ctx context.Context, arg DeleteBackupTargetByIDAndUserParams) error DeleteBanner(ctx context.Context, id uuid.UUID) error + DeleteCronJob(ctx context.Context, id uuid.UUID) error DeleteDatabaseServiceByID(ctx context.Context, id string) error DeleteDatabaseServiceByIDAndUser(ctx context.Context, arg DeleteDatabaseServiceByIDAndUserParams) error DeleteNotificationChannel(ctx context.Context, arg DeleteNotificationChannelParams) (int64, error) @@ -72,6 +73,7 @@ type Querier interface { DeleteUserInvite(ctx context.Context, id uuid.UUID) error DeleteUserTemplate(ctx context.Context, arg DeleteUserTemplateParams) (int64, error) FailDeployment(ctx context.Context, arg FailDeploymentParams) error + FinishCronExecution(ctx context.Context, arg FinishCronExecutionParams) error GetActiveAgentAuthTokenByHash(ctx context.Context, tokenHash string) (AgentAuthToken, error) GetAgent(ctx context.Context, id string) (NodeAgent, error) GetAgentByHostAndIP(ctx context.Context, arg GetAgentByHostAndIPParams) (NodeAgent, error) @@ -81,6 +83,11 @@ type Querier interface { GetCommandForAgent(ctx context.Context, arg GetCommandForAgentParams) (AgentCommand, error) GetContainer(ctx context.Context, id string) (ContainerInstance, error) GetContainerForAgent(ctx context.Context, arg GetContainerForAgentParams) (ContainerInstance, error) + GetCronJob(ctx context.Context, id uuid.UUID) (CronJob, error) + GetCronJobForTrigger(ctx context.Context, id uuid.UUID) (GetCronJobForTriggerRow, error) + GetCronJobOwner(ctx context.Context, id uuid.UUID) (uuid.UUID, error) + GetCronJobProjectID(ctx context.Context, id uuid.UUID) (uuid.UUID, error) + GetCronOwnerViaService(ctx context.Context, id uuid.UUID) (uuid.UUID, error) GetDatabaseBackupByIDAndDatabaseAndUser(ctx context.Context, arg GetDatabaseBackupByIDAndDatabaseAndUserParams) (DatabaseBackup, error) GetDatabaseServiceByID(ctx context.Context, id string) (DatabaseService, error) GetDatabaseServiceByIDAndUser(ctx context.Context, arg GetDatabaseServiceByIDAndUserParams) (DatabaseService, error) @@ -120,6 +127,8 @@ type Querier interface { GetUserInviteByTokenHash(ctx context.Context, tokenHash string) (UserInvite, error) GetUserTokenByHash(ctx context.Context, tokenHash string) (UserToken, error) InsertAgentHeartbeat(ctx context.Context, arg InsertAgentHeartbeatParams) error + InsertCronExecution(ctx context.Context, arg InsertCronExecutionParams) error + InsertCronJob(ctx context.Context, arg InsertCronJobParams) error InsertDeployment(ctx context.Context, arg InsertDeploymentParams) error InsertNotificationChannel(ctx context.Context, arg InsertNotificationChannelParams) (NotificationChannel, error) InsertPendingDeployment(ctx context.Context, arg InsertPendingDeploymentParams) error @@ -134,10 +143,12 @@ type Querier interface { ListBanners(ctx context.Context) ([]Banner, error) ListCommandsForAgent(ctx context.Context, nodeAgentID string) ([]AgentCommand, error) ListContainersForAgent(ctx context.Context, nodeAgentID string) ([]ContainerInstance, error) + ListCronExecutions(ctx context.Context, cronJobID uuid.UUID) ([]CronExecution, error) ListDatabaseBackupsByDatabaseAndUser(ctx context.Context, arg ListDatabaseBackupsByDatabaseAndUserParams) ([]DatabaseBackup, error) ListDatabaseServicesByProject(ctx context.Context, projectID uuid.NullUUID) ([]DatabaseService, error) ListDatabaseServicesByUser(ctx context.Context, userID string) ([]DatabaseService, error) ListDeploymentsForService(ctx context.Context, arg ListDeploymentsForServiceParams) ([]ListDeploymentsForServiceRow, error) + ListDueCronJobs(ctx context.Context) ([]ListDueCronJobsRow, error) ListDueDatabaseBackups(ctx context.Context) ([]ListDueDatabaseBackupsRow, error) ListEnabledOutboundWebhooks(ctx context.Context) ([]OutboundWebhook, error) ListExpiredPreviewsForUser(ctx context.Context, arg ListExpiredPreviewsForUserParams) ([]ListExpiredPreviewsForUserRow, error) @@ -215,6 +226,7 @@ type Querier interface { TouchAgentAuthToken(ctx context.Context, tokenHash string) error TouchServiceDomainChecked(ctx context.Context, id uuid.UUID) error TouchUserToken(ctx context.Context, id uuid.UUID) error + TrimCronExecutions(ctx context.Context, arg TrimCronExecutionsParams) error UpdateAgent(ctx context.Context, arg UpdateAgentParams) (NodeAgent, error) UpdateAgentHeartbeat(ctx context.Context, arg UpdateAgentHeartbeatParams) error UpdateBackupTargetByIDAndUser(ctx context.Context, arg UpdateBackupTargetByIDAndUserParams) error @@ -227,6 +239,7 @@ type Querier interface { // removed replica. $4 is the cumulative rx+tx byte counter. UpdateContainerStateByName(ctx context.Context, arg UpdateContainerStateByNameParams) error UpdateContainerStatus(ctx context.Context, arg UpdateContainerStatusParams) error + UpdateCronJobRun(ctx context.Context, arg UpdateCronJobRunParams) error UpdateDatabaseServiceNameAndPlanByIDAndUser(ctx context.Context, arg UpdateDatabaseServiceNameAndPlanByIDAndUserParams) error UpdateDatabaseServiceNameByIDAndUser(ctx context.Context, arg UpdateDatabaseServiceNameByIDAndUserParams) error UpdateDatabaseServicePlanByIDAndUser(ctx context.Context, arg UpdateDatabaseServicePlanByIDAndUserParams) error diff --git a/app/backend/sqlc/queries/cron.sql b/app/backend/sqlc/queries/cron.sql new file mode 100644 index 0000000..1a44dd5 --- /dev/null +++ b/app/backend/sqlc/queries/cron.sql @@ -0,0 +1,58 @@ +-- name: GetCronJobProjectID :one +SELECT project_id FROM cron_jobs WHERE id = $1; + +-- name: GetCronOwnerViaService :one +SELECT p.owner_id FROM projects p +JOIN services s ON s.project_id = p.id +WHERE s.id = $1; + +-- name: InsertCronJob :exec +INSERT INTO cron_jobs (id, project_id, service_id, name, schedule, command, timezone, enabled, next_run_at, retention, created_at, updated_at) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12); + +-- name: GetCronJob :one +SELECT id, project_id, service_id, name, schedule, command, timezone, + enabled, last_run_at, next_run_at, last_status, last_output, + retention, created_at, updated_at +FROM cron_jobs WHERE id = $1; + +-- name: GetCronJobOwner :one +SELECT p.owner_id FROM cron_jobs cj +JOIN projects p ON cj.project_id = p.id +WHERE cj.id = $1; + +-- name: DeleteCronJob :exec +DELETE FROM cron_jobs WHERE id = $1; + +-- name: ListCronExecutions :many +SELECT id, cron_job_id, started_at, finished_at, status, output, error +FROM cron_executions +WHERE cron_job_id = $1 +ORDER BY started_at DESC +LIMIT 100; + +-- name: GetCronJobForTrigger :one +SELECT cj.service_id, cj.command, cj.schedule, cj.timezone, cj.retention, p.owner_id +FROM cron_jobs cj +JOIN projects p ON cj.project_id = p.id +WHERE cj.id = $1; + +-- name: InsertCronExecution :exec +INSERT INTO cron_executions (id, cron_job_id, started_at, status) +VALUES ($1, $2, $3, $4); + +-- name: ListDueCronJobs :many +SELECT id, project_id, service_id, name, schedule, command, timezone, enabled, retention +FROM cron_jobs +WHERE enabled = TRUE AND next_run_at IS NOT NULL AND next_run_at <= NOW(); + +-- name: FinishCronExecution :exec +UPDATE cron_executions SET finished_at = $1, status = $2, output = $3, error = $4 WHERE id = $5; + +-- name: UpdateCronJobRun :exec +UPDATE cron_jobs SET last_run_at = $1, last_status = $2, last_output = $3, next_run_at = $4 WHERE id = $5; + +-- name: TrimCronExecutions :exec +DELETE FROM cron_executions WHERE cron_executions.cron_job_id = $1 AND cron_executions.id NOT IN ( + SELECT ce.id FROM cron_executions ce WHERE ce.cron_job_id = $1 + ORDER BY ce.started_at DESC LIMIT $2); diff --git a/app/backend/sqlc/schema.sql b/app/backend/sqlc/schema.sql index 373aa14..50d6ed6 100644 --- a/app/backend/sqlc/schema.sql +++ b/app/backend/sqlc/schema.sql @@ -459,3 +459,31 @@ CREATE TABLE instance_metrics ( created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), PRIMARY KEY (service_id, timestamp, instance_id) ); + +CREATE TABLE cron_jobs ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + project_id UUID NOT NULL REFERENCES projects(id) ON DELETE CASCADE, + service_id UUID NOT NULL REFERENCES services(id) ON DELETE CASCADE, + name VARCHAR(255) NOT NULL, + schedule VARCHAR(100) NOT NULL, + command TEXT NOT NULL, + timezone VARCHAR(50) DEFAULT 'UTC', + enabled BOOLEAN DEFAULT true, + last_run_at TIMESTAMP WITH TIME ZONE, + next_run_at TIMESTAMP WITH TIME ZONE, + last_status VARCHAR(50), + last_output TEXT, + retention INTEGER DEFAULT 30, + created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), + updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() +); + +CREATE TABLE cron_executions ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + cron_job_id UUID NOT NULL REFERENCES cron_jobs(id) ON DELETE CASCADE, + started_at TIMESTAMP WITH TIME ZONE NOT NULL, + finished_at TIMESTAMP WITH TIME ZONE, + status VARCHAR(50) DEFAULT 'pending', + output TEXT, + error TEXT +);