Task chaining + some refactoring

2025-12-11 14:08:52 +00:00 · 2019-02-15 22:10:02 -05:00
parent 07c0eca5aa
commit 6ca92bc0a7
31 changed files with 306 additions and 166 deletions
--- a/storage/auth.go
+++ b/storage/auth.go
@@ -131,7 +131,7 @@ func (database *Database) ManagerHasRoleOn(manager *Manager, projectId int64) Ma
 	db := database.getDB()

 	row := db.QueryRow(`SELECT role FROM manager_has_role_on_project 
-		WHERE project=$1 AND manager=$2`, projectId, manager.Id)
+		WHERE projectChange=$1 AND manager=$2`, projectId, manager.Id)

 	var role ManagerRole
 	err := row.Scan(&role)
--- a/storage/monitoring.go
+++ b/storage/monitoring.go
@@ -22,19 +22,19 @@ func (database *Database) MakeProjectSnapshots() {

 	insertRes, err := db.Exec(`
 		INSERT INTO project_monitoring_snapshot
-		  (project, new_task_count, failed_task_count, closed_task_count, worker_access_count,
+		  (projectChange, new_task_count, failed_task_count, closed_task_count, worker_access_count,
 		   awaiting_verification_task_count, timestamp)
 		SELECT id,
 			   (SELECT COUNT(*) FROM task 
 					LEFT JOIN worker_verifies_task wvt on task.id = wvt.task
-			   		WHERE task.project = project.id AND status = 1 AND wvt.task IS NULL),
-			   (SELECT COUNT(*) FROM task WHERE task.project = project.id AND status = 2),
+			   		WHERE task.projectChange = projectChange.id AND status = 1 AND wvt.task IS NULL),
+			   (SELECT COUNT(*) FROM task WHERE task.projectChange = projectChange.id AND status = 2),
 			   closed_task_count,
-			   (SELECT COUNT(*) FROM worker_has_access_to_project wa WHERE wa.project = project.id),
+			   (SELECT COUNT(*) FROM worker_has_access_to_project wa WHERE wa.projectChange = projectChange.id),
 			   (SELECT COUNT(*) FROM worker_verifies_task INNER JOIN task t on worker_verifies_task.task = t.id
-			  		WHERE t.project = project.id),
+			  		WHERE t.projectChange = projectChange.id),
 			   extract(epoch from now() at time zone 'utc')
-		FROM project`)
+		FROM projectChange`)
 	handleErr(err)
 	inserted, _ := insertRes.RowsAffected()

@@ -47,7 +47,7 @@ func (database *Database) MakeProjectSnapshots() {
 		"took":   time.Now().Sub(startTime),
 		"add":    inserted,
 		"remove": deleted,
-	}).Trace("Took project monitoring snapshot")
+	}).Trace("Took projectChange monitoring snapshot")
 }

 func (database *Database) GetMonitoringSnapshotsBetween(pid int64, from int, to int) (ss *[]ProjectMonitoringSnapshot) {
@@ -58,7 +58,7 @@ func (database *Database) GetMonitoringSnapshotsBetween(pid int64, from int, to

 	rows, err := db.Query(`SELECT new_task_count, failed_task_count, closed_task_count,
 		worker_access_count, awaiting_verification_task_count, timestamp FROM project_monitoring_snapshot 
-		WHERE project=$1 AND timestamp BETWEEN $2 AND $3 ORDER BY TIMESTAMP DESC `, pid, from, to)
+		WHERE projectChange=$1 AND timestamp BETWEEN $2 AND $3 ORDER BY TIMESTAMP DESC `, pid, from, to)
 	handleErr(err)

 	for rows.Next() {
@@ -89,7 +89,7 @@ func (database *Database) GetNMonitoringSnapshots(pid int64, count int) (ss *[]P

 	rows, err := db.Query(`SELECT new_task_count, failed_task_count, closed_task_count,
 		worker_access_count, awaiting_verification_task_count, timestamp FROM project_monitoring_snapshot 
-		WHERE project=$1 ORDER BY TIMESTAMP DESC LIMIT $2`, pid, count)
+		WHERE projectChange=$1 ORDER BY TIMESTAMP DESC LIMIT $2`, pid, count)
 	handleErr(err)

 	for rows.Next() {
--- a/storage/project.go
+++ b/storage/project.go
@@ -16,6 +16,7 @@ type Project struct {
 	Motd     string `json:"motd"`
 	Public   bool   `json:"public"`
 	Hidden   bool   `json:"hidden"`
+	Chain    int64  `json:"chain"`
 }

 type AssignedTasks struct {
@@ -25,34 +26,29 @@ type AssignedTasks struct {

 func (database *Database) SaveProject(project *Project) (int64, error) {
 	db := database.getDB()
-	id, projectErr := saveProject(project, db)

-	return id, projectErr
-}
-
-func saveProject(project *Project, db *sql.DB) (int64, error) {
-
-	row := db.QueryRow(`INSERT INTO project (name, git_repo, clone_url, version, priority, motd, public, hidden)
-		VALUES ($1,$2,$3,$4,$5,$6,$7,$8) RETURNING id`,
+	row := db.QueryRow(`INSERT INTO projectChange (name, git_repo, clone_url, version, priority,
+                     motd, public, hidden, chain)
+		VALUES ($1,$2,$3,$4,$5,$6,$7,$8,NULLIF($9, 0)) RETURNING id`,
 		project.Name, project.GitRepo, project.CloneUrl, project.Version, project.Priority, project.Motd,
-		project.Public, project.Hidden)
+		project.Public, project.Hidden, project.Chain)

 	var id int64
 	err := row.Scan(&id)

 	if err != nil {
 		logrus.WithError(err).WithFields(logrus.Fields{
-			"project": project,
-		}).Warn("Database.saveProject INSERT project ERROR")
+			"projectChange": project,
+		}).Warn("Database.saveProject INSERT projectChange ERROR")
 		return -1, err
 	}

 	project.Id = id

 	logrus.WithFields(logrus.Fields{
-		"id":      id,
-		"project": project,
-	}).Trace("Database.saveProject INSERT project")
+		"id":            id,
+		"projectChange": project,
+	}).Trace("Database.saveProject INSERT projectChange")

 	return id, nil
 }
@@ -60,52 +56,47 @@ func saveProject(project *Project, db *sql.DB) (int64, error) {
 func (database *Database) GetProject(id int64) *Project {

 	db := database.getDB()
-	project := getProject(id, db)
-	return project
-}
-
-func getProject(id int64, db *sql.DB) *Project {
-
-	row := db.QueryRow(`SELECT id, priority, name, clone_url, git_repo, version, motd, public, hidden
-		FROM project WHERE id=$1`, id)
+	row := db.QueryRow(`SELECT id, priority, name, clone_url, git_repo, version,
+       motd, public, hidden, COALESCE(chain, 0)
+		FROM projectChange WHERE id=$1`, id)

 	project, err := scanProject(row)
 	if err != nil {
 		logrus.WithError(err).WithFields(logrus.Fields{
 			"id": id,
-		}).Warn("Database.getProject SELECT project NOT FOUND")
+		}).Warn("Database.getProject SELECT projectChange NOT FOUND")
 		return nil
 	}

 	logrus.WithFields(logrus.Fields{
-		"id":      id,
-		"project": project,
-	}).Trace("Database.saveProject SELECT project")
+		"id":            id,
+		"projectChange": project,
+	}).Trace("Database.saveProject SELECT projectChange")

 	return project
 }

 func scanProject(row *sql.Row) (*Project, error) {

-	project := &Project{}
-	err := row.Scan(&project.Id, &project.Priority, &project.Name, &project.CloneUrl,
-		&project.GitRepo, &project.Version, &project.Motd, &project.Public, &project.Hidden)
+	p := &Project{}
+	err := row.Scan(&p.Id, &p.Priority, &p.Name, &p.CloneUrl, &p.GitRepo, &p.Version,
+		&p.Motd, &p.Public, &p.Hidden, &p.Chain)

-	return project, err
+	return p, err
 }

 func (database *Database) GetProjectWithRepoName(repoName string) *Project {

 	db := database.getDB()
-	row := db.QueryRow(`SELECT id, priority, name, clone_url, git_repo, version, motd, public, hidden 
-			FROM project WHERE LOWER(git_repo)=$1`,
+	row := db.QueryRow(`SELECT id, priority, name, clone_url, git_repo, version,
+       motd, public, hidden, COALESCE(chain, 0) FROM projectChange WHERE LOWER(git_repo)=$1`,
 		strings.ToLower(repoName))

 	project, err := scanProject(row)
 	if err != nil {
 		logrus.WithError(err).WithFields(logrus.Fields{
 			"repoName": repoName,
-		}).Warn("Database.getProjectWithRepoName SELECT project NOT FOUND")
+		}).Warn("Database.getProjectWithRepoName SELECT projectChange NOT FOUND")
 		return nil
 	}

@@ -116,10 +107,12 @@ func (database *Database) UpdateProject(project *Project) error {

 	db := database.getDB()

-	res, err := db.Exec(`UPDATE project 
-		SET (priority, name, clone_url, git_repo, version, motd, public, hidden) = ($1,$2,$3,$4,$5,$6,$7,$8) WHERE id=$9`,
+	res, err := db.Exec(`UPDATE projectChange 
+		SET (priority, name, clone_url, git_repo, version, motd, public, hidden, chain) =
+		  ($1,$2,$3,$4,$5,$6,$7,$8,NULLIF($9, 0))
+		WHERE id=$10`,
 		project.Priority, project.Name, project.CloneUrl, project.GitRepo, project.Version, project.Motd,
-		project.Public, project.Hidden, project.Id)
+		project.Public, project.Hidden, project.Chain, project.Id)
 	if err != nil {
 		return err
 	}
@@ -127,9 +120,9 @@ func (database *Database) UpdateProject(project *Project) error {
 	rowsAffected, _ := res.RowsAffected()

 	logrus.WithFields(logrus.Fields{
-		"project":      project,
-		"rowsAffected": rowsAffected,
-	}).Trace("Database.updateProject UPDATE project")
+		"projectChange": project,
+		"rowsAffected":  rowsAffected,
+	}).Trace("Database.updateProject UPDATE projectChange")

 	return nil
 }
@@ -142,15 +135,15 @@ func (database Database) GetAllProjects(workerId int64) *[]Project {
 	var err error
 	if workerId == 0 {
 		rows, err = db.Query(`SELECT 
-       	Id, priority, name, clone_url, git_repo, version, motd, public, hidden
-		FROM project
+       	Id, priority, name, clone_url, git_repo, version, motd, public, hidden, COALESCE(chain,0)
+		FROM projectChange
 		WHERE NOT hidden
 		ORDER BY name`)
 	} else {
 		rows, err = db.Query(`SELECT 
-       	Id, priority, name, clone_url, git_repo, version, motd, public, hidden
-		FROM project
-		LEFT JOIN worker_has_access_to_project whatp ON whatp.project = id
+       	Id, priority, name, clone_url, git_repo, version, motd, public, hidden, COALESCE(chain,0)
+		FROM projectChange
+		LEFT JOIN worker_has_access_to_project whatp ON whatp.projectChange = id
 		WHERE NOT hidden OR whatp.worker = $1
 		ORDER BY name`, workerId)
 	}
@@ -159,13 +152,14 @@ func (database Database) GetAllProjects(workerId int64) *[]Project {
 	for rows.Next() {
 		p := Project{}
 		err := rows.Scan(&p.Id, &p.Priority, &p.Name, &p.CloneUrl,
-			&p.GitRepo, &p.Version, &p.Motd, &p.Public, &p.Hidden)
+			&p.GitRepo, &p.Version, &p.Motd, &p.Public, &p.Hidden,
+			&p.Chain)
 		handleErr(err)
 		projects = append(projects, p)
 	}

 	logrus.WithFields(logrus.Fields{
-		"projects": projects,
+		"projects": len(projects),
 	}).Trace("Get all projects stats")

 	return &projects
@@ -177,7 +171,7 @@ func (database *Database) GetAssigneeStats(pid int64, count int64) *[]AssignedTa
 	assignees := make([]AssignedTasks, 0)

 	rows, err := db.Query(`SELECT worker.alias, COUNT(*) as wc FROM TASK
-  			LEFT JOIN worker ON TASK.assignee = worker.id WHERE project=$1 
+  			LEFT JOIN worker ON TASK.assignee = worker.id WHERE projectChange=$1 
 			GROUP BY worker.id ORDER BY wc LIMIT $2`, pid, count)
 	handleErr(err)

--- a/storage/task.go
+++ b/storage/task.go
@@ -9,7 +9,7 @@ import (
 type Task struct {
 	Id                int64      `json:"id"`
 	Priority          int64      `json:"priority"`
-	Project           *Project   `json:"project"`
+	Project           *Project   `json:"projectChange"`
 	Assignee          int64      `json:"assignee"`
 	Retries           int64      `json:"retries"`
 	MaxRetries        int64      `json:"max_retries"`
@@ -41,7 +41,7 @@ func (database *Database) SaveTask(task *Task, project int64, hash64 int64) erro

 	//TODO: For some reason it refuses to insert the 64-bit value unless I do that...
 	res, err := db.Exec(fmt.Sprintf(`
-	INSERT INTO task (project, max_retries, recipe, priority, max_assign_time, hash64,verification_count) 
+	INSERT INTO task (projectChange, max_retries, recipe, priority, max_assign_time, hash64,verification_count) 
 	VALUES ($1,$2,$3,$4,$5,NULLIF(%d, 0),$6)`, hash64),
 		project, task.MaxRetries, task.Recipe, task.Priority, task.MaxAssignTime, task.VerificationCount)
 	if err != nil {
@@ -73,14 +73,14 @@ func (database *Database) GetTask(worker *Worker) *Task {
 	(
 		SELECT task.id
 	FROM task
-	INNER JOIN project p on task.project = p.id
+	INNER JOIN projectChange project on task.projectChange = project.id
 	LEFT JOIN worker_verifies_task wvt on task.id = wvt.task AND wvt.worker=$1
 	WHERE assignee IS NULL AND task.status=1
-		AND (p.public OR EXISTS (
-		  SELECT 1 FROM worker_has_access_to_project a WHERE a.worker=$1 AND a.project=p.id
+		AND (project.public OR EXISTS (
+		  SELECT 1 FROM worker_has_access_to_project a WHERE a.worker=$1 AND a.projectChange=project.id
 		))
 		AND wvt.task IS NULL
-	ORDER BY p.priority DESC, task.priority DESC
+	ORDER BY project.priority DESC, task.priority DESC
 	LIMIT 1
 	)
 	RETURNING id`, worker.Id)
@@ -107,10 +107,10 @@ func (database *Database) GetTask(worker *Worker) *Task {
 func getTaskById(id int64, db *sql.DB) *Task {

 	row := db.QueryRow(`
-	SELECT task.id, task.priority, task.project, assignee, retries, max_retries,
-	        status, recipe, max_assign_time, assign_time, verification_count, p.priority, p.name,
-	       p.clone_url, p.git_repo, p.version, p.motd, p.public FROM task 
-	  INNER JOIN project p ON task.project = p.id
+	SELECT task.id, task.priority, task.projectChange, assignee, retries, max_retries,
+	        status, recipe, max_assign_time, assign_time, verification_count, project.priority, project.name,
+	       project.clone_url, project.git_repo, project.version, project.motd, project.public, COALESCE(project.chain,0) FROM task 
+	  INNER JOIN projectChange project ON task.projectChange = project.id
 	WHERE task.id=$1`, id)
 	project := &Project{}
 	task := &Task{}
@@ -119,7 +119,8 @@ func getTaskById(id int64, db *sql.DB) *Task {
 	err := row.Scan(&task.Id, &task.Priority, &project.Id, &task.Assignee,
 		&task.Retries, &task.MaxRetries, &task.Status, &task.Recipe, &task.MaxAssignTime,
 		&task.AssignTime, &task.VerificationCount, &project.Priority, &project.Name,
-		&project.CloneUrl, &project.GitRepo, &project.Version, &project.Motd, &project.Public)
+		&project.CloneUrl, &project.GitRepo, &project.Version, &project.Motd, &project.Public,
+		&project.Chain)
 	handleErr(err)

 	logrus.WithFields(logrus.Fields{
@@ -175,11 +176,11 @@ func (database *Database) GetTaskFromProject(worker *Worker, projectId int64) *T
 	(
 		SELECT task.id
 	FROM task
-	INNER JOIN project p on task.project = p.id
+	INNER JOIN projectChange project on task.projectChange = project.id
 	LEFT JOIN worker_verifies_task wvt on task.id = wvt.task AND wvt.worker=$1
-	WHERE assignee IS NULL AND p.id=$2 AND status=1
-		AND (p.public OR EXISTS (
-		  SELECT 1 FROM worker_has_access_to_project a WHERE a.worker=$1 AND a.project=$2
+	WHERE assignee IS NULL AND project.id=$2 AND status=1
+		AND (project.public OR EXISTS (
+		  SELECT 1 FROM worker_has_access_to_project a WHERE a.worker=$1 AND a.projectChange=$2
 		))
 		AND wvt.task IS NULL
 	ORDER BY task.priority DESC
--- a/storage/worker.go
+++ b/storage/worker.go
@@ -56,7 +56,7 @@ func (database *Database) GetWorker(id int64) *Worker {
 func (database *Database) GrantAccess(workerId int64, projectId int64) bool {

 	db := database.getDB()
-	res, err := db.Exec(`INSERT INTO worker_has_access_to_project (worker, project) VALUES ($1,$2)
+	res, err := db.Exec(`INSERT INTO worker_has_access_to_project (worker, projectChange) VALUES ($1,$2)
 		ON CONFLICT DO NOTHING`,
 		workerId, projectId)
 	if err != nil {
@@ -81,7 +81,7 @@ func (database *Database) GrantAccess(workerId int64, projectId int64) bool {
 func (database *Database) RemoveAccess(workerId int64, projectId int64) bool {

 	db := database.getDB()
-	res, err := db.Exec(`DELETE FROM worker_has_access_to_project WHERE worker=$1 AND project=$2`,
+	res, err := db.Exec(`DELETE FROM worker_has_access_to_project WHERE worker=$1 AND projectChange=$2`,
 		workerId, projectId)
 	handleErr(err)

@@ -118,8 +118,8 @@ func (database *Database) SaveAccessRequest(worker *Worker, projectId int64) boo
 	db := database.getDB()

 	res, err := db.Exec(`INSERT INTO worker_requests_access_to_project 
-  		SELECT $1, id FROM project WHERE id=$2 AND NOT project.public 
-		AND NOT EXISTS(SELECT * FROM worker_has_access_to_project WHERE worker=$1 AND project=$2)`,
+  		SELECT $1, id FROM projectChange WHERE id=$2 AND NOT projectChange.public 
+		AND NOT EXISTS(SELECT * FROM worker_has_access_to_project WHERE worker=$1 AND projectChange=$2)`,
 		worker.Id, projectId)
 	if err != nil {
 		return false
@@ -139,13 +139,13 @@ func (database *Database) AcceptAccessRequest(worker *Worker, projectId int64) b
 	db := database.getDB()

 	res, err := db.Exec(`DELETE FROM worker_requests_access_to_project 
-		WHERE worker=$1 AND project=$2`)
+		WHERE worker=$1 AND projectChange=$2`)
 	handleErr(err)

 	rowsAffected, _ := res.RowsAffected()
 	if rowsAffected == 1 {
 		_, err := db.Exec(`INSERT INTO worker_has_access_to_project 
-  			(worker, project) VALUES ($1,$2)`,
+  			(worker, projectChange) VALUES ($1,$2)`,
 			worker.Id, projectId)
 		handleErr(err)
 	}
@@ -162,7 +162,7 @@ func (database *Database) RejectAccessRequest(worker *Worker, projectId int64) b
 	db := database.getDB()

 	res, err := db.Exec(`DELETE FROM worker_requests_access_to_project 
-		  WHERE worker=$1 AND project=$2`, worker.Id, projectId)
+		  WHERE worker=$1 AND projectChange=$2`, worker.Id, projectId)
 	handleErr(err)

 	rowsAffected, _ := res.RowsAffected()
@@ -180,7 +180,7 @@ func (database *Database) GetAllAccessRequests(projectId int64) *[]Worker {

 	rows, err := db.Query(`SELECT id, alias, created FROM worker_requests_access_to_project
 		INNER JOIN worker w on worker_requests_access_to_project.worker = w.id
-		WHERE project=$1`,
+		WHERE projectChange=$1`,
 		projectId)
 	handleErr(err)