Task state machine

backend/ci_backends/github_actions.go

@@ -9,11 +9,11 @@ import (
 )
 
 type GithubActionCi struct {
-	client *github.Client
+	Client *github.Client
 }
 
-func (g *GithubActionCi) TriggerWorkflow(repoOwner string, repoName string, job models.DiggerJob, jobString string, commentId int64) error {
-	client := g.client
+func (g GithubActionCi) TriggerWorkflow(repoOwner string, repoName string, job models.DiggerJob, jobString string, commentId int64) error {
+	client := g.Client
 	log.Printf("TriggerGithubWorkflow: repoOwner: %v, repoName: %v, commentId: %v", repoOwner, repoName, commentId)
 	_, err := client.Actions.CreateWorkflowDispatchEventByFileName(context.Background(), repoOwner, repoName, "digger_workflow.yml", github.CreateWorkflowDispatchEventRequest{
 		Ref:    job.Batch.BranchName,

backend/go.sum

@@ -285,6 +285,8 @@ github.com/agext/levenshtein v1.2.1/go.mod h1:JEDfjyjHDjOF/1e4FlBE/PkbqA9OfWu2ki
 github.com/agext/levenshtein v1.2.2/go.mod h1:JEDfjyjHDjOF/1e4FlBE/PkbqA9OfWu2ki2W0IB5558=
 github.com/agext/levenshtein v1.2.3 h1:YB2fHEn0UJagG8T1rrWknE3ZQzWM06O8AMAatNn7lmo=
 github.com/agext/levenshtein v1.2.3/go.mod h1:JEDfjyjHDjOF/1e4FlBE/PkbqA9OfWu2ki2W0IB5558=
+github.com/alecthomas/kong v0.7.1 h1:azoTh0IOfwlAX3qN9sHWTxACE2oV8Bg2gAwBsMwDQY4=
+github.com/alecthomas/kong v0.7.1/go.mod h1:n1iCIO2xS46oE8ZfYCNDqdR0b0wZNrXAIAqro/2132U=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=

backend/migrations/20240403155357_drop_dup_idx.sql

@@ -0,0 +1,5 @@
+# drop the duplicate index to fix the next migration of renaming
+DROP INDEX "public"."idx_digger_job_id";
+DROP INDEX "idx_digger_run_queues_deleted_at";
+DROP INDEX "idx_digger_run_queue_project_id";
+DROP INDEX "idx_digger_run_queue_run_id";

backend/migrations/20240403155456.sql

@@ -0,0 +1,24 @@
+-- Create index "idx_digger_job_id" to table: "digger_jobs"
+CREATE INDEX "idx_digger_job_id" ON "public"."digger_jobs" ("batch_id");
+-- Create index "idx_digger_run_batch_id" to table: "digger_run_stages"
+CREATE INDEX "idx_digger_run_batch_id" ON "public"."digger_run_stages" ("batch_id");
+-- Create "digger_run_queue_items" table
+CREATE TABLE "public"."digger_run_queue_items" (
+  "id" bigserial NOT NULL,
+  "created_at" timestamptz NULL,
+  "updated_at" timestamptz NULL,
+  "deleted_at" timestamptz NULL,
+  "project_id" bigint NULL,
+  "digger_run_id" bigint NULL,
+  PRIMARY KEY ("id"),
+  CONSTRAINT "fk_digger_run_queue_items_digger_run" FOREIGN KEY ("digger_run_id") REFERENCES "public"."digger_runs" ("id") ON UPDATE NO ACTION ON DELETE NO ACTION,
+  CONSTRAINT "fk_digger_run_queue_items_project" FOREIGN KEY ("project_id") REFERENCES "public"."projects" ("id") ON UPDATE NO ACTION ON DELETE NO ACTION
+);
+-- Create index "idx_digger_run_queue_items_deleted_at" to table: "digger_run_queue_items"
+CREATE INDEX "idx_digger_run_queue_items_deleted_at" ON "public"."digger_run_queue_items" ("deleted_at");
+-- Create index "idx_digger_run_queue_project_id" to table: "digger_run_queue_items"
+CREATE INDEX "idx_digger_run_queue_project_id" ON "public"."digger_run_queue_items" ("project_id");
+-- Create index "idx_digger_run_queue_run_id" to table: "digger_run_queue_items"
+CREATE INDEX "idx_digger_run_queue_run_id" ON "public"."digger_run_queue_items" ("digger_run_id");
+-- Drop "digger_run_queues" table
+DROP TABLE "public"."digger_run_queues";

backend/migrations/atlas.sum

@@ -1,4 +1,4 @@
-h1:XKQNc29tdcumex/EDFrmq1Xlus0chqk3B9DYku62Hfs=
+h1:AwUIRwmx9E+aRp469t6izImEEgfQt6IHwtAKXumn4jo=
 20231227132525.sql h1:43xn7XC0GoJsCnXIMczGXWis9d504FAWi4F1gViTIcw=
 20240115170600.sql h1:IW8fF/8vc40+eWqP/xDK+R4K9jHJ9QBSGO6rN9LtfSA=
 20240116123649.sql h1:R1JlUIgxxF6Cyob9HdtMqiKmx/BfnsctTl5rvOqssQw=
@@ -9,3 +9,5 @@ h1:XKQNc29tdcumex/EDFrmq1Xlus0chqk3B9DYku62Hfs=
 20240329100957.sql h1:6IHn/Se6FwdmipMDPAPF0yChNNCuwxrEt4rgn+0gkLQ=
 20240329114422.sql h1:chXvrIUFNud2SdbRClWSCKXZ4MrMu0mpgE08Bou3pgk=
 20240402110915.sql h1:bG2Dvbzm3ZvFa29Feb0Bwj6KtAtZy1Vyuje6yV31msQ=
+20240403155357_drop_dup_idx.sql h1:6LyRtGfutHQompownriYYrq8us+Cdj4FTgWa7VPsXFA=
+20240403155456.sql h1:XJgyne416JMAV4xHA3IweHZos0ULrjFEJBqhWFjGNho=

backend/models/runs.go

@@ -11,9 +11,14 @@ type DiggerRunStatus string
 
 const (
 	RunQueued          DiggerRunStatus = "Queued"
+	RunPendingPlan     DiggerRunStatus = "Pending Plan"
+	RunPlanning        DiggerRunStatus = "Running Plan"
+	RunPendingApproval DiggerRunStatus = "Pending Approval"
+	RunApproved        DiggerRunStatus = "Approved"
+	RunPendingApply    DiggerRunStatus = "Pending Apply"
+	RunApplying        DiggerRunStatus = "Running Apply"
 	RunSucceeded       DiggerRunStatus = "Succeeded"
 	RunFailed          DiggerRunStatus = "Failed"
-	RunPendingApproval DiggerRunStatus = "Pending Approval"
 )
 
 type RunType string
@@ -23,7 +28,7 @@ const (
 	PlanOnly     RunType = "Plan Only"
 )
 
-type DiggerRunQueue struct {
+type DiggerRunQueueItem struct {
 	gorm.Model
 	ProjectId   uint `gorm:"index:idx_digger_run_queue_project_id"`
 	Project     *Project
@@ -52,7 +57,7 @@ type DiggerRunStage struct {
 	Run     *DiggerRun
 	RunID   uint `gorm:"index:idx_digger_run_stage_id"`
 	Batch   *DiggerBatch
-	BatchID *string `gorm:"index:idx_digger_job_id"`
+	BatchID *string `gorm:"index:idx_digger_run_batch_id"`
 }
 
 type SerializedRunStage struct {

backend/models/storage.go

@@ -594,6 +594,15 @@ func (db *Database) CreateDiggerBatch(githubInstallationId int64, repoOwner stri
 	return batch, nil
 }
 
+func (db *Database) UpdateDiggerBatch(batch *DiggerBatch) error {
+	result := db.GormDB.Save(batch)
+	if result.Error != nil {
+		return result.Error
+	}
+	log.Printf("batch %v has been updated successfully\n", batch.ID)
+	return nil
+}
+
 func (db *Database) UpdateBatchStatus(batch *DiggerBatch) error {
 	if batch.Status == scheduler.BatchJobInvalidated || batch.Status == scheduler.BatchJobFailed || batch.Status == scheduler.BatchJobSucceeded {
 		return nil
@@ -646,6 +655,59 @@ func (db *Database) CreateDiggerJob(batchId uuid.UUID, serializedJob []byte, wor
 	return job, nil
 }
 
+func (db *Database) CreateDiggerRun(Triggertype string, PrNumber int, Status DiggerRunStatus, CommitId string, DiggerConfig string, GithubInstallationId int64, RepoId uint, ProjectID uint, RunType RunType) (*DiggerRun, error) {
+	dr := &DiggerRun{
+		Triggertype:          Triggertype,
+		PrNumber:             &PrNumber,
+		Status:               Status,
+		CommitId:             CommitId,
+		DiggerConfig:         DiggerConfig,
+		GithubInstallationId: GithubInstallationId,
+		RepoId:               RepoId,
+		ProjectID:            ProjectID,
+		RunType:              RunType,
+	}
+	result := db.GormDB.Save(dr)
+	if result.Error != nil {
+		log.Printf("Failed to create DiggerRun: %v, error: %v\n", dr.ID, result.Error)
+		return nil, result.Error
+	}
+	log.Printf("DiggerRun %v, has been created successfully\n", dr.ID)
+	return dr, nil
+}
+
+func (db *Database) GetDiggerRun(id uint) (*DiggerRun, error) {
+	dr := &DiggerRun{}
+	result := db.GormDB.Where("id=? ", id).Find(dr)
+	if result.Error != nil {
+		return nil, result.Error
+	}
+	return dr, nil
+}
+
+func (db *Database) CreateDiggerRunQueueItem(projectId uint, diggeRrunId uint) (*DiggerRunQueueItem, error) {
+	drq := &DiggerRunQueueItem{
+		ProjectId:   projectId,
+		DiggerRunId: diggeRrunId,
+	}
+	result := db.GormDB.Save(drq)
+	if result.Error != nil {
+		log.Printf("Failed to create DiggerRunQueueItem: %v, error: %v\n", drq.ID, result.Error)
+		return nil, result.Error
+	}
+	log.Printf("DiggerRunQueueItem %v, has been created successfully\n", drq.ID)
+	return drq, nil
+}
+
+func (db *Database) GetDiggerRunQueueItem(id uint) (*DiggerRunQueueItem, error) {
+	dr := &DiggerRunQueueItem{}
+	result := db.GormDB.Preload("Project").Preload("DiggerRun").Where("id=? ", id).Find(dr)
+	if result.Error != nil {
+		return nil, result.Error
+	}
+	return dr, nil
+}
+
 func (db *Database) GetDiggerJobFromRunStage(stage DiggerRunStage) (*DiggerJob, error) {
 	job := &DiggerJob{}
 	result := db.GormDB.Take(job, "batch_id = ?", stage.BatchID)
@@ -659,8 +721,27 @@ func (db *Database) GetDiggerJobFromRunStage(stage DiggerRunStage) (*DiggerJob,
 	return job, nil
 }
 
-func (db *Database) GetFirstRunQueueForEveryProject() ([]DiggerRunQueue, error) {
-	var runqueues []DiggerRunQueue
+func (db *Database) UpdateDiggerRun(diggerRun *DiggerRun) error {
+	result := db.GormDB.Save(diggerRun)
+	if result.Error != nil {
+		return result.Error
+	}
+	log.Printf("diggerRun %v has been updated successfully\n", diggerRun.ID)
+	return nil
+}
+
+func (db *Database) DequeueRunItem(queue *DiggerRunQueueItem) error {
+	log.Printf("DiggerRunQueueItem Deleting: %v", queue.ID)
+	result := db.GormDB.Delete(queue)
+	if result.Error != nil {
+		return result.Error
+	}
+	log.Printf("diggerRunQueueItem %v has been deleted successfully\n")
+	return nil
+}
+
+func (db *Database) GetFirstRunQueueForEveryProject() ([]DiggerRunQueueItem, error) {
+	var runqueues []DiggerRunQueueItem
 	query := `WITH RankedRuns AS (
   SELECT
     digger_run_queues.digger_run_id,
@@ -690,11 +771,11 @@ WHERE
 	}
 
 	// 2. Preload Project and DiggerRun for every DiggerrunQueue item (front of queue)
-	var runqueuesWithData []DiggerRunQueue
-	projectIds := lo.Map(runqueues, func(run DiggerRunQueue, index int) uint {
+	var runqueuesWithData []DiggerRunQueueItem
+	projectIds := lo.Map(runqueues, func(run DiggerRunQueueItem, index int) uint {
 		return run.ProjectId
 	})
-	diggerRunIds := lo.Map(runqueues, func(run DiggerRunQueue, index int) uint {
+	diggerRunIds := lo.Map(runqueues, func(run DiggerRunQueueItem, index int) uint {
 		return run.DiggerRunId
 	})
 

backend/tasks/runs.go

@@ -0,0 +1,108 @@
+package main
+
+import (
+	"github.com/diggerhq/digger/backend/ci_backends"
+	"github.com/diggerhq/digger/backend/models"
+	orchestrator_scheduler "github.com/diggerhq/digger/libs/orchestrator/scheduler"
+	"log"
+)
+
+func RunQueuesStateMachine(queueItem *models.DiggerRunQueueItem, CIBackend ci_backends.CiBackend) {
+	dr := queueItem.DiggerRun
+	switch queueItem.DiggerRun.Status {
+	case models.RunQueued:
+		// trigger plan workflow (trigger the batch)
+		// .....
+		// change status to RunPendingPlan
+		log.Printf("Updating run queueItem item to planning state")
+		dr.Status = models.RunPlanning
+		err := models.DB.UpdateDiggerRun(&dr)
+		if err != nil {
+			log.Printf("ERROR: Failed to update Digger Run for queueID: %v [%v %v]", queueItem.ID, queueItem.DiggerRunId, queueItem.ProjectId)
+		}
+	case models.RunPlanning:
+		// Check the status of the batch
+		batchStatus := orchestrator_scheduler.BatchJobSucceeded
+		approvalRequired := true
+
+		// if failed then go straight to failed
+		if batchStatus == orchestrator_scheduler.BatchJobFailed {
+			dr.Status = models.RunFailed
+			err := models.DB.UpdateDiggerRun(&dr)
+			if err != nil {
+				log.Printf("ERROR: Failed to update Digger Run for queueID: %v [%v %v]", queueItem.ID, queueItem.DiggerRunId, queueItem.ProjectId)
+			}
+			err = models.DB.DequeueRunItem(queueItem)
+			if err != nil {
+				log.Printf("ERROR: Failed to delete queueItem item: %v [%v %v]", queueItem.ID, queueItem.DiggerRunId, queueItem.ProjectId)
+			}
+		}
+
+		// if successful then
+		if batchStatus == orchestrator_scheduler.BatchJobSucceeded && approvalRequired {
+			dr.Status = models.RunPendingApproval
+			err := models.DB.UpdateDiggerRun(&dr)
+			if err != nil {
+				log.Printf("ERROR: Failed to update Digger Run for queueID: %v [%v %v]", queueItem.ID, queueItem.DiggerRunId, queueItem.ProjectId)
+			}
+		} else {
+			dr.Status = models.RunApproved
+			err := models.DB.UpdateDiggerRun(&dr)
+			if err != nil {
+				log.Printf("ERROR: Failed to update Digger Run for queueID: %v [%v %v]", queueItem.ID, queueItem.DiggerRunId, queueItem.ProjectId)
+			}
+		}
+
+	case models.RunPendingApproval:
+		// do nothing
+	case models.RunApproved:
+		// trigger apply stage workflow
+		// ...
+		dr.Status = models.RunApplying
+		err := models.DB.UpdateDiggerRun(&dr)
+		if err != nil {
+			log.Printf("ERROR: Failed to update Digger Run for queueID: %v [%v %v]", queueItem.ID, queueItem.DiggerRunId, queueItem.ProjectId)
+		}
+
+	case models.RunApplying:
+		// Check the status of the batch
+		batchStatus := orchestrator_scheduler.BatchJobSucceeded
+
+		// if failed then go straight to failed
+		if batchStatus == orchestrator_scheduler.BatchJobFailed {
+			dr.Status = models.RunFailed
+			err := models.DB.UpdateDiggerRun(&dr)
+			if err != nil {
+				log.Printf("ERROR: Failed to update Digger Run for queueID: %v [%v %v]", queueItem.ID, queueItem.DiggerRunId, queueItem.ProjectId)
+			}
+			err = models.DB.DequeueRunItem(queueItem)
+			if err != nil {
+				log.Printf("ERROR: Failed to delete queueItem item: %v [%v %v]", queueItem.ID, queueItem.DiggerRunId, queueItem.ProjectId)
+			}
+		}
+
+		// if successful then
+		if batchStatus == orchestrator_scheduler.BatchJobSucceeded {
+			dr.Status = models.RunSucceeded
+			err := models.DB.UpdateDiggerRun(&dr)
+			if err != nil {
+				log.Printf("ERROR: Failed to update Digger Run for queueID: %v [%v %v]", queueItem.ID, queueItem.DiggerRunId, queueItem.ProjectId)
+			}
+		}
+
+	case models.RunSucceeded:
+		// dequeue
+		err := models.DB.DequeueRunItem(queueItem)
+		if err != nil {
+			log.Printf("ERROR: Failed to delete queueItem item: %v [%v %v]", queueItem.ID, queueItem.DiggerRunId, queueItem.ProjectId)
+		}
+	case models.RunFailed:
+		// dequeue
+		err := models.DB.DequeueRunItem(queueItem)
+		if err != nil {
+			log.Printf("ERROR: Failed to delete queueItem item: %v [%v %v]", queueItem.ID, queueItem.DiggerRunId, queueItem.ProjectId)
+		}
+	default:
+		log.Printf("WARN: Recieived unknown DiggerRunStatus: %v", queueItem.DiggerRun.Status)
+	}
+}