mirror of
https://github.com/Dvorinka/Containr.git
synced 2026-10-28 07:41:27 +00:00
Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
23d085c98f | ||
|
|
269e031761 | ||
|
|
ce71b5af24 |
No files matched your search
+142
-118
@@ -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,
|
||||
)
|
||||
_ = q.TrimCronExecutions(context.Background(), sqlcdb.TrimCronExecutionsParams{
|
||||
CronJobID: jid,
|
||||
Limit: int32(retention),
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -573,4 +573,29 @@ func TestRuntimeLiveInvariants(t *testing.T) {
|
||||
if row.Replicas != 3 {
|
||||
t.Fatalf("replicas: %+v", row)
|
||||
}
|
||||
|
||||
// Sleeper reads: candidate sweep + domain/status text lookups.
|
||||
_, err = db.Exec(`UPDATE services SET sleep_enabled = true, domain = 'rt.example.com', status = 'running' WHERE id = $1`, serviceID)
|
||||
must(err)
|
||||
cands, err := q.ListSleepCandidates(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("sleep candidates: %v", err)
|
||||
}
|
||||
var mine *sqlcdb.ListSleepCandidatesRow
|
||||
for i := range cands {
|
||||
if cands[i].ID == serviceID {
|
||||
mine = &cands[i]
|
||||
}
|
||||
}
|
||||
if mine == nil || mine.SleepIdleMinutes != 15 {
|
||||
t.Fatalf("candidate missing/idle wrong: %+v", mine)
|
||||
}
|
||||
dom, err := q.GetServiceDomainText(ctx, serviceID)
|
||||
if err != nil || dom != "rt.example.com" {
|
||||
t.Fatalf("domain text: %v %q", err, dom)
|
||||
}
|
||||
st, err := q.GetServiceStatusText(ctx, serviceID)
|
||||
if err != nil || st != "running" {
|
||||
t.Fatalf("status text: %v %q", err, st)
|
||||
}
|
||||
}
|
||||
@@ -86,20 +86,17 @@ type sleepCandidate struct {
|
||||
}
|
||||
|
||||
func sweepSleeping(ctx context.Context, db *database.DB, dc *docker.Client, engine *deployment.DeploymentEngine, apiPort int) {
|
||||
rows, err := db.QueryContext(ctx,
|
||||
`SELECT id, project_id, name, COALESCE(status,''), COALESCE(sleep_idle_minutes,15)
|
||||
FROM services WHERE sleep_enabled AND COALESCE(status,'') IN ('running','deployed','degraded','starting')`)
|
||||
sweepRows, err := sqlcdb.New(db.DB).ListSleepCandidates(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
var candidates []sleepCandidate
|
||||
for rows.Next() {
|
||||
var s sleepCandidate
|
||||
if err := rows.Scan(&s.ID, &s.ProjectID, &s.Name, &s.Status, &s.IdleMinutes); err == nil {
|
||||
candidates = append(candidates, s)
|
||||
for _, r := range sweepRows {
|
||||
candidates = append(candidates, sleepCandidate{
|
||||
ID: r.ID, ProjectID: r.ProjectID, Name: r.Name,
|
||||
Status: r.Status, IdleMinutes: int(r.SleepIdleMinutes),
|
||||
})
|
||||
}
|
||||
}
|
||||
rows.Close()
|
||||
|
||||
q := sqlcdb.New(db.DB)
|
||||
for _, svc := range candidates {
|
||||
@@ -187,11 +184,10 @@ func putToSleepRemote(ctx context.Context, db *database.DB, q *sqlcdb.Queries, s
|
||||
tracker.mu.Lock()
|
||||
delete(tracker.activity, svc.ID)
|
||||
tracker.mu.Unlock()
|
||||
_, _ = db.Exec(`UPDATE services SET status = 'sleeping', updated_at = $1 WHERE id = $2`, time.Now(), svc.ID)
|
||||
setServiceStatusNow(db, svc.ID, "sleeping")
|
||||
log.Printf("sleep: %s (%s) remote replicas removed after idle timeout", svc.Name, svc.ID)
|
||||
|
||||
var domain string
|
||||
_ = db.QueryRow(`SELECT COALESCE(domain,'') FROM services WHERE id = $1`, svc.ID).Scan(&domain)
|
||||
domain, _ := q.GetServiceDomainText(ctx, svc.ID)
|
||||
domains := serviceDomainNames(db, svc.ID, domain)
|
||||
if len(domains) > 0 {
|
||||
writeWakeIngress(svc.ID.String(), domains, apiPort)
|
||||
@@ -250,11 +246,10 @@ func putToSleep(ctx context.Context, db *database.DB, dc *docker.Client, engine
|
||||
tracker.mu.Lock()
|
||||
delete(tracker.activity, svc.ID)
|
||||
tracker.mu.Unlock()
|
||||
_, _ = db.Exec(`UPDATE services SET status = 'sleeping', updated_at = $1 WHERE id = $2`, time.Now(), svc.ID)
|
||||
setServiceStatusNow(db, svc.ID, "sleeping")
|
||||
log.Printf("sleep: %s (%s) scaled to zero after idle timeout", svc.Name, svc.ID)
|
||||
|
||||
var domain string
|
||||
_ = db.QueryRow(`SELECT COALESCE(domain,'') FROM services WHERE id = $1`, svc.ID).Scan(&domain)
|
||||
domain, _ := sqlcdb.New(db.DB).GetServiceDomainText(ctx, svc.ID)
|
||||
domains := serviceDomainNames(db, svc.ID, domain)
|
||||
if len(domains) == 0 {
|
||||
return
|
||||
@@ -444,7 +439,9 @@ func pruneWakeContainers(ctx context.Context, db *database.DB, dc *docker.Client
|
||||
}
|
||||
}
|
||||
var status string
|
||||
_ = db.QueryRow(`SELECT COALESCE(status,'') FROM services WHERE id = $1`, svcID).Scan(&status)
|
||||
if sid, err := uuid.Parse(svcID); err == nil {
|
||||
status, _ = sqlcdb.New(db.DB).GetServiceStatusText(ctx, sid)
|
||||
}
|
||||
if live > 0 || status == "stopped" || status == "" {
|
||||
_ = dc.RemoveContainer(ctx, c.ID, true)
|
||||
}
|
||||
@@ -491,34 +488,11 @@ func triggerServiceWake(c *gin.Context) (int, gin.H) {
|
||||
if err != nil {
|
||||
return http.StatusBadRequest, gin.H{"error": "invalid service id"}
|
||||
}
|
||||
var service Service
|
||||
err = db.QueryRow(
|
||||
`SELECT s.id, s.project_id, s.name,
|
||||
COALESCE(s.type, s.service_type, ''), COALESCE(s.status, ''),
|
||||
COALESCE(s.image, s.image_name, ''), COALESCE(s.command, s.start_command, ''),
|
||||
COALESCE(s.environment, ''), COALESCE(s.git_repo, s.source_url, ''),
|
||||
COALESCE(s.git_branch, ''), COALESCE(s.build_path, ''),
|
||||
COALESCE(s.cpu, ''), COALESCE(s.memory, ''),
|
||||
COALESCE(s.replicas, 1), COALESCE(s.port, 0),
|
||||
COALESCE(s.domain, ''), COALESCE(s.healthcheck_path, ''),
|
||||
COALESCE(s.restart_policy, 'unless-stopped'),
|
||||
COALESCE(s.builder, 'auto'), COALESCE(s.cpu_reserve, ''),
|
||||
COALESCE(s.memory_reserve, ''), COALESCE(s.static_build_cmd, ''),
|
||||
COALESCE(s.static_dir, ''),
|
||||
s.created_at, s.updated_at
|
||||
FROM services s WHERE s.id = $1`, id,
|
||||
).Scan(
|
||||
&service.ID, &service.ProjectID, &service.Name, &service.Type, &service.Status,
|
||||
&service.Image, &service.Command, &service.Environment, &service.GitRepo,
|
||||
&service.GitBranch, &service.BuildPath, &service.CPU, &service.Memory,
|
||||
&service.Replicas, &service.Port, &service.Domain, &service.HealthCheckPath,
|
||||
&service.RestartPolicy, &service.Builder, &service.CPUReserve,
|
||||
&service.MemoryReserve, &service.StaticBuildCmd, &service.StaticDir,
|
||||
&service.CreatedAt, &service.UpdatedAt,
|
||||
)
|
||||
row, err := sqlcdb.New(db.DB).GetServiceRuntimeWithOwner(context.Background(), id)
|
||||
if err != nil {
|
||||
return http.StatusNotFound, gin.H{"error": "service not found"}
|
||||
}
|
||||
service := serviceFromRuntimeRow(row)
|
||||
if service.Status == "running" || service.Status == "deployed" {
|
||||
return http.StatusOK, gin.H{"status": service.Status}
|
||||
}
|
||||
@@ -528,13 +502,13 @@ func triggerServiceWake(c *gin.Context) (int, gin.H) {
|
||||
spec, err := serviceRuntimeSpec(db, service)
|
||||
if err != nil {
|
||||
log.Printf("wake: runtime spec for %s failed: %v", service.ID, err)
|
||||
_, _ = db.Exec(`UPDATE services SET status = 'failed', updated_at = $1 WHERE id = $2`, time.Now(), service.ID)
|
||||
setServiceStatusNow(db, service.ID, "failed")
|
||||
return
|
||||
}
|
||||
state, err := engine.ReconcileService(ctx, spec)
|
||||
if err != nil {
|
||||
log.Printf("wake: reconcile for %s failed: %v", service.ID, err)
|
||||
_, _ = db.Exec(`UPDATE services SET status = 'failed', updated_at = $1 WHERE id = $2`, time.Now(), service.ID)
|
||||
setServiceStatusNow(db, service.ID, "failed")
|
||||
return
|
||||
}
|
||||
status := "running"
|
||||
@@ -542,7 +516,7 @@ func triggerServiceWake(c *gin.Context) (int, gin.H) {
|
||||
status = state.Status
|
||||
persistPublishedPort(db, service.ID, state.Ports)
|
||||
}
|
||||
_, _ = db.Exec(`UPDATE services SET status = $1, updated_at = $2 WHERE id = $3`, status, time.Now(), service.ID)
|
||||
setServiceStatusNow(db, service.ID, status)
|
||||
}()
|
||||
return http.StatusAccepted, gin.H{"status": "waking"}
|
||||
}
|
||||
@@ -563,8 +537,6 @@ func handleServiceSleep(c *gin.Context) {
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
var port int
|
||||
_ = db.QueryRow(`SELECT COALESCE(port,0) FROM services WHERE id = $1`, service.ID).Scan(&port)
|
||||
svc := sleepCandidate{
|
||||
ID: service.ID,
|
||||
ProjectID: service.ProjectID,
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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"`
|
||||
@@ -373,6 +401,8 @@ type Service struct {
|
||||
StaticDir string `json:"static_dir"`
|
||||
NodeID sql.NullString `json:"node_id"`
|
||||
Spread bool `json:"spread"`
|
||||
SleepEnabled bool `json:"sleep_enabled"`
|
||||
SleepIdleMinutes int32 `json:"sleep_idle_minutes"`
|
||||
PlacementTags json.RawMessage `json:"placement_tags"`
|
||||
TraefikLabels json.RawMessage `json:"traefik_labels"`
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
@@ -107,6 +114,7 @@ type Querier interface {
|
||||
GetServiceAccess(ctx context.Context, id uuid.UUID) (GetServiceAccessRow, error)
|
||||
GetServiceDomainDefault(ctx context.Context, serviceID uuid.UUID) (string, error)
|
||||
GetServiceDomainFirst(ctx context.Context, serviceID uuid.UUID) (string, error)
|
||||
GetServiceDomainText(ctx context.Context, id uuid.UUID) (string, error)
|
||||
GetServiceEnvironment(ctx context.Context, id uuid.UUID) (string, error)
|
||||
GetServiceForDeploy(ctx context.Context, id uuid.UUID) (GetServiceForDeployRow, error)
|
||||
GetServiceForDeployWithOwner(ctx context.Context, id uuid.UUID) (GetServiceForDeployWithOwnerRow, error)
|
||||
@@ -114,12 +122,15 @@ type Querier interface {
|
||||
GetServicePort(ctx context.Context, id uuid.UUID) (int32, error)
|
||||
GetServicePublishedPort(ctx context.Context, id uuid.UUID) (int32, error)
|
||||
GetServiceRuntimeWithOwner(ctx context.Context, id uuid.UUID) (GetServiceRuntimeWithOwnerRow, error)
|
||||
GetServiceStatusText(ctx context.Context, id uuid.UUID) (string, error)
|
||||
GetServiceTemplateByID(ctx context.Context, id string) (ServiceTemplate, error)
|
||||
GetServiceTypeBrief(ctx context.Context, arg GetServiceTypeBriefParams) (GetServiceTypeBriefRow, error)
|
||||
GetServiceVariableValue(ctx context.Context, arg GetServiceVariableValueParams) (string, error)
|
||||
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 +145,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)
|
||||
@@ -172,6 +185,7 @@ type Querier interface {
|
||||
ListServiceTemplatesForUser(ctx context.Context, ownerID uuid.NullUUID) ([]ServiceTemplate, error)
|
||||
ListServiceVariableValues(ctx context.Context, serviceID uuid.UUID) ([]ListServiceVariableValuesRow, error)
|
||||
ListServiceVariablesWithSecret(ctx context.Context, serviceID uuid.UUID) ([]ListServiceVariablesWithSecretRow, error)
|
||||
ListSleepCandidates(ctx context.Context) ([]ListSleepCandidatesRow, error)
|
||||
ListSweepablePreviews(ctx context.Context) ([]ListSweepablePreviewsRow, error)
|
||||
ListUserInvites(ctx context.Context) ([]UserInvite, error)
|
||||
ListUserTokens(ctx context.Context, userID uuid.UUID) ([]UserToken, error)
|
||||
@@ -215,6 +229,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 +242,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
|
||||
|
||||
@@ -25,6 +25,17 @@ func (q *Queries) GetLastDeployedImage(ctx context.Context, serviceID uuid.UUID)
|
||||
return image, err
|
||||
}
|
||||
|
||||
const getServiceDomainText = `-- name: GetServiceDomainText :one
|
||||
SELECT COALESCE(domain,'') AS domain FROM services WHERE id = $1
|
||||
`
|
||||
|
||||
func (q *Queries) GetServiceDomainText(ctx context.Context, id uuid.UUID) (string, error) {
|
||||
row := q.db.QueryRowContext(ctx, getServiceDomainText, id)
|
||||
var domain string
|
||||
err := row.Scan(&domain)
|
||||
return domain, err
|
||||
}
|
||||
|
||||
const getServicePort = `-- name: GetServicePort :one
|
||||
SELECT COALESCE(port, 0) AS port FROM services WHERE id = $1
|
||||
`
|
||||
@@ -118,6 +129,17 @@ func (q *Queries) GetServiceRuntimeWithOwner(ctx context.Context, id uuid.UUID)
|
||||
return i, err
|
||||
}
|
||||
|
||||
const getServiceStatusText = `-- name: GetServiceStatusText :one
|
||||
SELECT COALESCE(status,'') AS status FROM services WHERE id = $1
|
||||
`
|
||||
|
||||
func (q *Queries) GetServiceStatusText(ctx context.Context, id uuid.UUID) (string, error) {
|
||||
row := q.db.QueryRowContext(ctx, getServiceStatusText, id)
|
||||
var status string
|
||||
err := row.Scan(&status)
|
||||
return status, err
|
||||
}
|
||||
|
||||
const getServiceVariableValue = `-- name: GetServiceVariableValue :one
|
||||
SELECT value FROM environment_variables WHERE service_id = $1 AND key = $2
|
||||
`
|
||||
@@ -259,6 +281,50 @@ func (q *Queries) ListServiceVariablesWithSecret(ctx context.Context, serviceID
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const listSleepCandidates = `-- name: ListSleepCandidates :many
|
||||
SELECT id, project_id, name, COALESCE(status,'') AS status,
|
||||
COALESCE(sleep_idle_minutes, 15) AS sleep_idle_minutes
|
||||
FROM services
|
||||
WHERE sleep_enabled AND COALESCE(status,'') IN ('running','deployed','degraded','starting')
|
||||
`
|
||||
|
||||
type ListSleepCandidatesRow struct {
|
||||
ID uuid.UUID `json:"id"`
|
||||
ProjectID uuid.UUID `json:"project_id"`
|
||||
Name string `json:"name"`
|
||||
Status string `json:"status"`
|
||||
SleepIdleMinutes int32 `json:"sleep_idle_minutes"`
|
||||
}
|
||||
|
||||
func (q *Queries) ListSleepCandidates(ctx context.Context) ([]ListSleepCandidatesRow, error) {
|
||||
rows, err := q.db.QueryContext(ctx, listSleepCandidates)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
items := []ListSleepCandidatesRow{}
|
||||
for rows.Next() {
|
||||
var i ListSleepCandidatesRow
|
||||
if err := rows.Scan(
|
||||
&i.ID,
|
||||
&i.ProjectID,
|
||||
&i.Name,
|
||||
&i.Status,
|
||||
&i.SleepIdleMinutes,
|
||||
); 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 setServicePublishedPort = `-- name: SetServicePublishedPort :exec
|
||||
UPDATE services SET published_port = $1 WHERE id = $2 AND published_port <> $1
|
||||
`
|
||||
|
||||
@@ -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);
|
||||
@@ -45,3 +45,15 @@ UPDATE services SET published_port = $1 WHERE id = $2 AND published_port <> $1;
|
||||
|
||||
-- name: SetServiceReplicas :exec
|
||||
UPDATE services SET replicas = $1, updated_at = $2 WHERE id = $3;
|
||||
|
||||
-- name: ListSleepCandidates :many
|
||||
SELECT id, project_id, name, COALESCE(status,'') AS status,
|
||||
COALESCE(sleep_idle_minutes, 15) AS sleep_idle_minutes
|
||||
FROM services
|
||||
WHERE sleep_enabled AND COALESCE(status,'') IN ('running','deployed','degraded','starting');
|
||||
|
||||
-- name: GetServiceDomainText :one
|
||||
SELECT COALESCE(domain,'') AS domain FROM services WHERE id = $1;
|
||||
|
||||
-- name: GetServiceStatusText :one
|
||||
SELECT COALESCE(status,'') AS status FROM services WHERE id = $1;
|
||||
@@ -69,6 +69,8 @@ CREATE TABLE services (
|
||||
static_dir VARCHAR(255) NOT NULL DEFAULT '',
|
||||
node_id VARCHAR(255),
|
||||
spread BOOLEAN NOT NULL DEFAULT false,
|
||||
sleep_enabled BOOLEAN NOT NULL DEFAULT false,
|
||||
sleep_idle_minutes INTEGER NOT NULL DEFAULT 15,
|
||||
placement_tags JSONB NOT NULL DEFAULT '[]',
|
||||
traefik_labels JSONB NOT NULL DEFAULT '{}'
|
||||
);
|
||||
@@ -459,3 +461,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
|
||||
);
|
||||
Reference in new issue
Block a user