feat (api): new SSE handler for last Update (#865)

engine/api/admin.go

@@ -5,8 +5,8 @@ import (
 
 	"github.com/go-gorp/gorp"
 
-	"github.com/ovh/cds/engine/api/cache"
 	"github.com/ovh/cds/engine/api/businesscontext"
+	"github.com/ovh/cds/engine/api/cache"
 	"github.com/ovh/cds/sdk/log"
 )
 

engine/api/application.go

@@ -341,6 +341,7 @@ func addApplicationHandler(w http.ResponseWriter, r *http.Request, db *gorp.DbMa
 }
 
 func deleteApplicationHandler(w http.ResponseWriter, r *http.Request, db *gorp.DbMap, c *businesscontext.Ctx) error {
+
 	nb, errNb := pipeline.CountBuildingPipelineByApplication(db, c.Application.ID)
 	if errNb != nil {
 		log.Warning("deleteApplicationHandler> Cannot count pipeline build for application %d: %s\n", c.Application.ID, errNb)
@@ -359,14 +360,16 @@ func deleteApplicationHandler(w http.ResponseWriter, r *http.Request, db *gorp.D
 	}
 	defer tx.Rollback()
 
-	err = application.DeleteApplication(tx, c.Application.ID)
-	if err != nil {
+	if err := application.DeleteApplication(tx, c.Application.ID); err != nil {
 		log.Warning("deleteApplicationHandler> Cannot delete application: %s\n", err)
 		return err
 	}
 
-	err = tx.Commit()
-	if err != nil {
+	if err := project.UpdateLastModified(tx, c.User, c.Project); err != nil {
+		return sdk.WrapError(err, "deleteApplicationHandler> Cannot update project last modified date")
+	}
+
+	if err := tx.Commit(); err != nil {
 		log.Warning("deleteApplicationHandler> Cannot commit transaction: %s\n", err)
 		return err
 	}

engine/api/application/application_delete.go

@@ -93,15 +93,5 @@ func DeleteApplication(db gorp.SqlExecutor, applicationID int64) error {
 		log.Warning("DeleteApplication> Cannot delete application: %s\n", err)
 		return err
 	}
-
-	// Update project
-	query = `
-		UPDATE project
-		SET last_modified = current_timestamp
-		WHERE id IN (
-			select project_id from application where id = $1
-		)
-	`
-	_, err = db.Exec(query, applicationID)
-	return err
+	return nil
 }

engine/api/application/application_group.go

@@ -130,7 +130,7 @@ func AddGroup(db gorp.SqlExecutor, proj *sdk.Project, a *sdk.Application, u *sdk
 					return sdk.WrapError(err, "AddGroup> Cannot add group %s in pipeline %s", g.Name, p.Pipeline.Name)
 				}
 
-				if err := pipeline.UpdateLastModified(db, u, &p.Pipeline); err != nil {
+				if err := pipeline.UpdatePipelineLastModified(db, proj, &p.Pipeline, u); err != nil {
 					return sdk.WrapError(err, "AddGroup> Cannot update pipeline %s", p.Pipeline.Name)
 				}
 			}

engine/api/application/application_variable.go

@@ -366,10 +366,7 @@ func DeleteAllVariable(db gorp.SqlExecutor, applicationID int64) error {
 	if err != nil {
 		return err
 	}
-
-	query = "UPDATE application SET last_modified = current_timestamp WHERE id=$1"
-	_, err = db.Exec(query, applicationID)
-	return err
+	return nil
 }
 
 // AddKeyPairToApplication generate a ssh key pair and add them as application variables

engine/api/application/dao.go

@@ -2,6 +2,7 @@ package application
 
 import (
 	"database/sql"
+	"encoding/json"
 	"fmt"
 	"time"
 
@@ -196,7 +197,7 @@ func Update(db gorp.SqlExecutor, app *sdk.Application, u *sdk.User) error {
 // UpdateLastModified Update last_modified column in application table
 func UpdateLastModified(db gorp.SqlExecutor, app *sdk.Application, u *sdk.User) error {
 	query := `
-		UPDATE application SET last_modified=current_timestamp WHERE id = $1 RETURNING last_modified
+		UPDATE application SET last_modified = current_timestamp WHERE id = $1 RETURNING last_modified
 	`
 	var lastModified time.Time
 	err := db.QueryRow(query, app.ID).Scan(&lastModified)
@@ -210,6 +211,18 @@ func UpdateLastModified(db gorp.SqlExecutor, app *sdk.Application, u *sdk.User)
 			Username:     u.Username,
 			LastModified: lastModified.Unix(),
 		}, 0)
+
+		updates := sdk.LastModification{
+			Key:          app.ProjectKey,
+			Name:         app.Name,
+			LastModified: lastModified.Unix(),
+			Username:     u.Username,
+			Type:         sdk.ApplicationLastModificationType,
+		}
+		b, errP := json.Marshal(updates)
+		if errP == nil {
+			cache.Publish("lastUpdates", string(b))
+		}
 	}
 
 	return sdk.WrapError(err, "application.UpdateLastModified %s(%d)", app.Name, app.ID)

engine/api/application_pipeline_notif_test.go

@@ -102,7 +102,7 @@ func testApplicationPipelineNotifBoilerPlate(t *testing.T, f func(*testing.T, *g
 		ProjectID:  proj.ID,
 	}
 	t.Logf("Insert Pipeline %s for Project %s", pip.Name, proj.Name)
-	err := pipeline.InsertPipeline(db, pip, u)
+	err := pipeline.InsertPipeline(db, proj, pip, u)
 	test.NoError(t, err)
 
 	//Insert Application
@@ -607,7 +607,7 @@ func Test_addNotificationsHandler(t *testing.T) {
 		Type:      "build",
 		ProjectID: p.ID,
 	}
-	err = pipeline.InsertPipeline(db, pip, u)
+	err = pipeline.InsertPipeline(db, p, pip, u)
 	test.NoError(t, err)
 
 	_, err = application.AttachPipeline(db, app.ID, pip.ID)

engine/api/application_pipeline_test.go

@@ -39,7 +39,7 @@ func Test_attachPipelinesToApplicationHandler(t *testing.T) {
 		ProjectID:  proj.ID,
 	}
 
-	if err := pipeline.InsertPipeline(db, pip, u); err != nil {
+	if err := pipeline.InsertPipeline(db, proj, pip, u); err != nil {
 		t.Fatal(err)
 	}
 
@@ -51,7 +51,7 @@ func Test_attachPipelinesToApplicationHandler(t *testing.T) {
 		ProjectID:  proj.ID,
 	}
 
-	if err := pipeline.InsertPipeline(db, pip2, u); err != nil {
+	if err := pipeline.InsertPipeline(db, proj, pip2, u); err != nil {
 		t.Fatal(err)
 	}
 

engine/api/application_test.go

@@ -41,7 +41,7 @@ func TestGetApplicationWithTriggersHandler(t *testing.T) {
 		ProjectKey: proj.Key,
 		ProjectID:  proj.ID,
 	}
-	if err := pipeline.InsertPipeline(db, pip1, u); err != nil {
+	if err := pipeline.InsertPipeline(db, proj, pip1, u); err != nil {
 		t.Fatal(err)
 	}
 
@@ -53,7 +53,7 @@ func TestGetApplicationWithTriggersHandler(t *testing.T) {
 		ProjectKey: proj.Key,
 		ProjectID:  proj.ID,
 	}
-	if err := pipeline.InsertPipeline(db, pip2, u); err != nil {
+	if err := pipeline.InsertPipeline(db, proj, pip2, u); err != nil {
 		t.Fatal(err)
 	}
 
@@ -65,7 +65,7 @@ func TestGetApplicationWithTriggersHandler(t *testing.T) {
 		ProjectKey: proj.Key,
 		ProjectID:  proj.ID,
 	}
-	if err := pipeline.InsertPipeline(db, pip3, u); err != nil {
+	if err := pipeline.InsertPipeline(db, proj, pip3, u); err != nil {
 		t.Fatal(err)
 	}
 	// 6. Create application

engine/api/build_test.go

@@ -46,7 +46,7 @@ func Test_updateStepStatusHandler(t *testing.T) {
 		ProjectID:  proj.ID,
 	}
 
-	if err := pipeline.InsertPipeline(db, pip, u); err != nil {
+	if err := pipeline.InsertPipeline(db, proj, pip, u); err != nil {
 		t.Fatal(err)
 	}
 
@@ -142,7 +142,7 @@ func Test_addSpawnInfosPipelineBuildJobHandler(t *testing.T) {
 		ProjectID:  proj.ID,
 	}
 
-	if err := pipeline.InsertPipeline(db, pip, u); err != nil {
+	if err := pipeline.InsertPipeline(db, proj, pip, u); err != nil {
 		t.Fatal(err)
 	}
 

engine/api/cache/cache.go

@@ -3,6 +3,7 @@ package cache
 import (
 	"container/list"
 	"context"
+	"fmt"
 	"strings"
 	"sync"
 
@@ -12,6 +13,11 @@ import (
 //Status : local ok redis
 var Status string
 
+// PubSub represents a subscriber
+type PubSub interface {
+	Unsubscribe(channels ...string) error
+}
+
 //Key make a key as expected
 func Key(args ...string) string {
 	return strings.Join(args, ":")
@@ -28,6 +34,9 @@ type Store interface {
 	Dequeue(queueName string, value interface{})
 	DequeueWithContext(c context.Context, queueName string, value interface{})
 	QueueLen(queueName string) int
+	Publish(queueName string, value interface{})
+	Subscribe(queueName string) PubSub
+	GetMessageFromSubscription(c context.Context, pb PubSub) (string, error)
 }
 
 //Initialize the global cache in memory, or redis
@@ -128,3 +137,27 @@ func QueueLen(queueName string) int {
 	}
 	return s.QueueLen(queueName)
 }
+
+// Publish a message on a channel
+func Publish(queueName string, value interface{}) {
+	if s == nil {
+		return
+	}
+	s.Publish(queueName, value)
+}
+
+// Subscribe to a channel
+func Subscribe(queueName string) PubSub {
+	if s == nil {
+		return nil
+	}
+	return s.Subscribe(queueName)
+}
+
+// GetMessageFromSubscription Get a message from a subscription
+func GetMessageFromSubscription(c context.Context, pb PubSub) (string, error) {
+	if s == nil {
+		return "", fmt.Errorf("Cache > Client store is nil")
+	}
+	return s.GetMessageFromSubscription(c, pb)
+}