chore: add pipeline status

.gitignore

@@ -101,8 +101,8 @@ mysql-persistence/
 # Dependency directories (remove the comment below to include it)
 # vendor/
 
-# built binary
-pipeline-backend
+# local protogen-go
+protogen-go
 
 # air related files and folders
 .air.toml

configs/config.yaml

@@ -15,7 +15,7 @@ database:
   host: localhost
   port: 5432
   name: pipeline
-  version: 4
+  version: 6
   timezone: Etc/UTC
   pool:
     idleconnections: 5

go.mod

@@ -9,7 +9,7 @@ require (
 	github.com/golang/mock v1.6.0
 	github.com/grpc-ecosystem/go-grpc-middleware v1.3.0
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.7.3
-	github.com/instill-ai/protogen-go v0.1.5-alpha
+	github.com/instill-ai/protogen-go v0.1.5-alpha.0.20220406002712-68dad8357630
 	github.com/instill-ai/vdp v0.1.3-alpha.0.20220320033117-c3e0db375b45
 	github.com/instill-ai/x v0.1.0-alpha
 	github.com/knadh/koanf v1.4.0

go.sum

@@ -596,8 +596,8 @@ github.com/imdario/mergo v0.3.11/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH
 github.com/imdario/mergo v0.3.12/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/instill-ai/protogen-go v0.1.3-alpha/go.mod h1:q2Pq4P0AY/59RGibT4nSDnOsA4wD4XhLueFRoGYNBjk=
-github.com/instill-ai/protogen-go v0.1.5-alpha h1:rmovaQhNaTFGuZxXPWLpYX5E6jqS/akWWaJ+3yAx0Xo=
-github.com/instill-ai/protogen-go v0.1.5-alpha/go.mod h1:q2Pq4P0AY/59RGibT4nSDnOsA4wD4XhLueFRoGYNBjk=
+github.com/instill-ai/protogen-go v0.1.5-alpha.0.20220406002712-68dad8357630 h1:MrBHKXx1h26Doikv6FaRqpQYnsVSN8Qm4L3W7PMetBM=
+github.com/instill-ai/protogen-go v0.1.5-alpha.0.20220406002712-68dad8357630/go.mod h1:q2Pq4P0AY/59RGibT4nSDnOsA4wD4XhLueFRoGYNBjk=
 github.com/instill-ai/vdp v0.1.3-alpha.0.20220320033117-c3e0db375b45 h1:MureMU/JCh3C0QwBVVgv1HsIPRcCN+wxqpg9vnvfutU=
 github.com/instill-ai/vdp v0.1.3-alpha.0.20220320033117-c3e0db375b45/go.mod h1:xcFgWKOHkqqkKMUkc5ftoD/5/hMSVR3qNhnnHxWzxYQ=
 github.com/instill-ai/x v0.1.0-alpha h1:cXzOGkHr+u1XdB8Pn6VHYb+hZX621mvlz7qBxZ29S5Y=

integration-test/rest.js

@@ -17,6 +17,7 @@ const model_name = pipelineConstants.detectionModel.name;
 const det_model = open(`${__ENV.TEST_FOLDER_ABS_PATH}/integration-test/data/dummy-det-model.zip`, "b");
 
 export let options = {
+  setupTimeout: '300s',
   insecureSkipTLSVerify: true,
   thresholds: {
     checks: ["rate == 1.0"],
@@ -94,7 +95,7 @@ export default function (data) {
       {
         name: randomString(100),
         description: randomString(512),
-        active: true,
+        status: "STATUS_ACTIVE",
       },
       pipelineConstants.detectionRecipe
     );
@@ -189,7 +190,7 @@ export default function (data) {
     let updatePipelineEntity = Object.assign(
       {
         description: randomString(512),
-        active: true,
+        status: "STATUS_ACTIVE",
       },
     );
     group("Pipelines API: Update a pipeline", () => {

internal/db/db.go

@@ -7,6 +7,7 @@ import (
 
 	"gorm.io/driver/postgres"
 	"gorm.io/gorm"
+	"gorm.io/gorm/schema"
 
 	configs "github.com/instill-ai/pipeline-backend/configs"
 )
@@ -31,6 +32,9 @@ func GetConnection() *gorm.DB {
 			PreferSimpleProtocol: true, // disables implicit prepared statement usage
 		}), &gorm.Config{
 			QueryFields: true, // QueryFields mode will select by all fields’ name for current model
+			NamingStrategy: schema.NamingStrategy{
+				SingularTable: true,
+			},
 		})
 
 		if err != nil {

internal/db/migration/000001_init.down.sql

@@ -1,9 +1,5 @@
 BEGIN;
 
-ALTER TABLE "pipeline_history" DROP CONSTRAINT IF EXISTS "pipeline_history_fk_pipeline_id";
-
-DROP TABLE IF EXISTS "pipeline_history";
-
-DROP TABLE IF EXISTS "pipelines";
+DROP TABLE IF EXISTS pipelines;
 
 COMMIT;

internal/db/migration/000001_init.up.sql

@@ -1,35 +1,23 @@
 BEGIN;
 
-CREATE TABLE IF NOT EXISTS "pipelines" (
-  "id" SERIAL PRIMARY KEY,
-  "ext_id" varchar(20) NOT NULL,
-  "name" varchar(256) NOT NULL,
-  "description" text,
-  "enabled" boolean NOT NULL DEFAULT (false),
-  "created_at" timestamp DEFAULT CURRENT_TIMESTAMP NOT NULL,
-  "updated_at" timestamp DEFAULT CURRENT_TIMESTAMP NOT NULL,
-  "deleted_at" timestamp,
-  "recipe" JSONB,
-  "crontab" varchar(13)
+CREATE TABLE IF NOT EXISTS pipelines (
+  id SERIAL PRIMARY KEY,
+  ext_id varchar(20) NOT NULL,
+  name varchar(256) NOT NULL,
+  description text,
+  enabled boolean NOT NULL DEFAULT (false),
+  created_at timestamp DEFAULT CURRENT_TIMESTAMP NOT NULL,
+  updated_at timestamp DEFAULT CURRENT_TIMESTAMP NOT NULL,
+  deleted_at timestamp,
+  recipe JSONB,
+  crontab varchar(13)
 );
 
-CREATE TABLE IF NOT EXISTS "pipeline_history" (
-  "pipeline_id" int NOT NULL,
-  "recipe" JSONB NOT NULL,
-  "version" int NOT NULL,
-  "created_at" timestamp DEFAULT CURRENT_TIMESTAMP NOT NULL,
-  PRIMARY KEY ("pipeline_id", "version")
-);
-
-COMMENT ON COLUMN "pipelines"."ext_id" IS 'the hash of the id';
-COMMENT ON COLUMN "pipelines"."name" IS 'name of this pipeline';
-COMMENT ON COLUMN "pipelines"."description" IS 'description of this pipeline';
-COMMENT ON COLUMN "pipelines"."enabled" IS 'activate/deactivate pipeline';
-COMMENT ON COLUMN "pipelines"."recipe" IS 'describe what the pipeline looks like';
-COMMENT ON COLUMN "pipelines"."crontab" IS 'the 6 * crontab format';
-
-COMMENT ON COLUMN "pipeline_history"."recipe" IS 'describe what the pipeline looks like';
-
-ALTER TABLE "pipeline_history" ADD CONSTRAINT "pipeline_history_fk_pipeline_id" FOREIGN KEY ("pipeline_id") REFERENCES "pipelines"("id");
+COMMENT ON COLUMN pipelines.ext_id IS 'the hash of the id';
+COMMENT ON COLUMN pipelines.name IS 'name of this pipeline';
+COMMENT ON COLUMN pipelines.description IS 'description of this pipeline';
+COMMENT ON COLUMN pipelines.enabled IS 'activate/deactivate pipeline';
+COMMENT ON COLUMN pipelines.recipe IS 'pipeline configration';
+COMMENT ON COLUMN pipelines.crontab IS 'the 6 * crontab format';
 
 COMMIT;

internal/db/migration/000002_create_column_creator_id.down.sql

@@ -1,5 +1,5 @@
 BEGIN;
 
-ALTER TABLE "pipelines" DROP "creator_id";
+ALTER TABLE pipelines DROP creator_id;
 
 COMMIT;

internal/db/migration/000002_create_column_creator_id.up.sql

@@ -1,5 +1,5 @@
 BEGIN;
 
-ALTER TABLE "pipelines" ADD "creator_id" varchar(36) NOT NULL;
+ALTER TABLE pipelines ADD creator_id varchar(36) NOT NULL;
 
 COMMIT;

internal/db/migration/000003_change_enabled_column_name.down.sql

@@ -1,5 +1,5 @@
 BEGIN;
 
-ALTER TABLE "pipelines" RENAME COLUMN "active" TO "enabled";
+ALTER TABLE pipelines RENAME COLUMN active TO enabled;
 
 COMMIT;

internal/db/migration/000003_change_enabled_column_name.up.sql

@@ -1,5 +1,5 @@
 BEGIN;
 
-ALTER TABLE "pipelines" RENAME COLUMN "enabled" TO "active";
+ALTER TABLE pipelines RENAME COLUMN enabled TO active;
 
 COMMIT;

internal/db/migration/000004_human_readable_name.down.sql

@@ -1,21 +1,19 @@
 BEGIN;
 
-ALTER TABLE "pipelines" ALTER COLUMN "id" SET DATA TYPE INT;
+ALTER TABLE pipelines DROP CONSTRAINT unique_name_namespace;
 
-ALTER TABLE "pipelines" DROP CONSTRAINT "unique_name_namespace";
+ALTER TABLE pipelines DROP COLUMN IF EXISTS namespace;
 
-ALTER TABLE "pipelines" DROP COLUMN IF EXISTS "namespace";
+ALTER TABLE pipelines ADD ext_id varchar(20) NOT NULL DEFAULT '000004';
+COMMENT ON COLUMN pipelines.ext_id IS 'the hash of the id';
 
-ALTER TABLE "pipelines" ADD "ext_id" varchar(20) NOT NULL DEFAULT '000004';
-COMMENT ON COLUMN "pipelines"."ext_id" IS 'the hash of the id';
+ALTER TABLE pipelines ADD creator_id varchar(36) NOT NULL DEFAULT '000004';
 
-ALTER TABLE "pipelines" ADD "creator_id" varchar(36) NOT NULL DEFAULT '000004';
+UPDATE pipelines
+SET ext_id = migration.ext_id, creator_id = migration.creator_id
+FROM pipelines_migration AS migration
+WHERE pipelines.id = migration.id;
 
-UPDATE "pipelines"
-SET "ext_id" = bak."ext_id", "creator_id" = bak."creator_id"
-FROM "pipelines_bak" AS bak
-WHERE "pipelines"."id" = bak."id";
-
-DROP TABLE IF EXISTS "pipelines_bak";
+DROP TABLE IF EXISTS pipelines_migration;
 
 COMMIT;

internal/db/migration/000004_human_readable_name.up.sql

@@ -1,15 +1,13 @@
 BEGIN;
 
-CREATE TABLE IF NOT EXISTS "pipelines_bak" AS TABLE "pipelines";
+CREATE TABLE IF NOT EXISTS pipelines_migration AS TABLE pipelines;
 
-ALTER TABLE "pipelines" DROP COLUMN IF EXISTS "ext_id";
-ALTER TABLE "pipelines" DROP COLUMN IF EXISTS "creator_id";
+ALTER TABLE pipelines DROP COLUMN IF EXISTS ext_id;
+ALTER TABLE pipelines DROP COLUMN IF EXISTS creator_id;
 
-ALTER TABLE "pipelines" ADD "namespace" varchar(39) NOT NULL DEFAULT 'undefined';
-COMMENT ON COLUMN "pipelines"."namespace" IS 'a set of pipelines';
+ALTER TABLE pipelines ADD namespace varchar(39) NOT NULL DEFAULT 'undefined';
+COMMENT ON COLUMN pipelines.namespace IS 'namespace in which the pipeline belongs to';
 
-ALTER TABLE "pipelines" ADD CONSTRAINT "unique_name_namespace" UNIQUE ("name", "deleted_at", "namespace");
-
-ALTER TABLE "pipelines" ALTER COLUMN "id" SET DATA TYPE BIGINT;
+ALTER TABLE pipelines ADD CONSTRAINT unique_name_namespace UNIQUE (name, deleted_at, namespace);
 
 COMMIT;

internal/db/migration/000005_rename_pipeline_table_to_singular.down.sql

@@ -0,0 +1,7 @@
+BEGIN;
+
+ALTER TABLE pipeline RENAME TO pipelines;
+
+ALTER TABLE pipelines RENAME CONSTRAINT pipeline_pkey TO pipelines_pkey;
+
+COMMIT;

internal/db/migration/000005_rename_pipeline_table_to_singular.up.sql

@@ -0,0 +1,7 @@
+BEGIN;
+
+ALTER TABLE pipelines RENAME TO pipeline;
+
+ALTER TABLE pipeline RENAME CONSTRAINT pipelines_pkey TO pipeline_pkey;
+
+COMMIT;

internal/db/migration/000006_add_pipeline_status_column.down.sql

@@ -0,0 +1,14 @@
+BEGIN;
+
+ALTER TABLE pipeline ALTER COLUMN status DROP DEFAULT;
+
+ALTER TABLE pipeline ALTER COLUMN status TYPE boolean
+USING CASE WHEN status='STATUS_INACTIVE'::valid_status THEN FALSE ELSE TRUE END;
+
+ALTER TABLE pipeline ALTER COLUMN status SET DEFAULT FALSE;
+
+ALTER TABLE pipeline RENAME COLUMN status TO active;
+
+DROP TYPE valid_status;
+
+COMMIT;

internal/db/migration/000006_add_pipeline_status_column.up.sql

@@ -0,0 +1,16 @@
+BEGIN;
+
+CREATE TYPE valid_status AS ENUM ('STATUS_INACTIVE', 'STATUS_ACTIVE', 'STATUS_ERROR');
+
+ALTER TABLE pipeline RENAME COLUMN active TO status;
+
+ALTER TABLE pipeline ALTER COLUMN status DROP DEFAULT;
+
+ALTER TABLE pipeline ALTER COLUMN status TYPE valid_status
+USING CASE WHEN status=FALSE THEN 'STATUS_INACTIVE'::valid_status ELSE 'STATUS_ACTIVE'::valid_status END;
+
+ALTER TABLE pipeline ALTER COLUMN status SET DEFAULT 'STATUS_INACTIVE'::valid_status;
+
+COMMENT ON COLUMN pipeline.status IS 'pipeline status';
+
+COMMIT;

pkg/datamodel/datamodel.go

@@ -1,30 +1,22 @@
 package datamodel
 
 import (
-	"time"
+	"database/sql/driver"
+	"errors"
 
 	"gorm.io/gorm"
 )
 
 type Pipeline struct {
-	Id uint64
+	gorm.Model
 
 	// the name of this Instill Pipeline
 	Name string
 
 	// the more detail of this Instill Pipeline
 	Description string
 
-	Active bool `gorm:"type:tinyint"`
-
-	// the time when entity created
-	CreatedAt time.Time `gorm:"type:timestamp"`
-
-	// the time when entity has been updated
-	UpdatedAt time.Time `gorm:"type:timestamp"`
-
-	// the time when entity has been deleted
-	DeletedAt gorm.DeletedAt
+	Status ValidStatus `sql:"type:valid_status"`
 
 	Recipe *Recipe `gorm:"type:json"`
 
@@ -50,3 +42,27 @@ type TriggerPipelineContent struct {
 	Base64 string `json:"base64,omitempty"`
 	Chunk  []byte `json:"chunk,omitempty"`
 }
+
+type ValidStatus string
+
+const (
+	StatusInactive ValidStatus = "STATUS_INACTIVE"
+	StatusActive   ValidStatus = "STATUS_ACTIVE"
+	StatusError    ValidStatus = "STATUS_ERROR"
+)
+
+func (p *ValidStatus) Scan(value interface{}) error {
+	switch v := value.(type) {
+	case string:
+		*p = ValidStatus(v)
+	case []byte:
+		*p = ValidStatus(v)
+	default:
+		return errors.New("Incompatible type for ValidStatus")
+	}
+	return nil
+}
+
+func (p ValidStatus) Value() (driver.Value, error) {
+	return string(p), nil
+}

pkg/datamodel/recipe.go

@@ -23,7 +23,7 @@ type Destination struct {
 }
 
 type Model struct {
-	Name    string `json:"model_name,omitempty"`
+	Name    string `json:"model,omitempty"`
 	Version uint64 `json:"version,omitempty"`
 }
 

pkg/handler/handler.go

@@ -75,7 +75,7 @@ func (s *handler) CreatePipeline(ctx context.Context, in *pipelinePB.CreatePipel
 	entity := datamodel.Pipeline{
 		Name:        in.Name,
 		Description: in.Description,
-		Active:      in.Active,
+		Status:      datamodel.ValidStatus(in.Status.String()),
 		Namespace:   username,
 	}
 	if in.Recipe != nil {
@@ -124,10 +124,10 @@ func (s *handler) ListPipeline(ctx context.Context, in *pipelinePB.ListPipelineR
 	var nextCursor uint64
 	for _, pipeline := range pipelines {
 		resp.Pipelines = append(resp.Pipelines, marshalPipeline(&pipeline))
-		nextCursor = pipeline.Id
+		nextCursor = uint64(pipeline.ID)
 	}
 
-	if min != nextCursor {
+	if uint64(min) != nextCursor {
 		resp.NextPageToken = s.paginateTocken.Encode(nextCursor)
 	}
 
@@ -168,8 +168,8 @@ func (s *handler) UpdatePipeline(ctx context.Context, in *pipelinePB.UpdatePipel
 			switch field {
 			case "description":
 				entity.Description = in.PipelinePatch.Description
-			case "active":
-				entity.Active = in.PipelinePatch.Active
+			case "status":
+				entity.Status = datamodel.ValidStatus(in.PipelinePatch.Status.String())
 			}
 			if strings.Contains(field, "recipe") {
 				entity.Recipe = unmarshalRecipe(in.PipelinePatch.Recipe)

pkg/handler/marshal.go

@@ -41,10 +41,10 @@ func marshalRecipe(recipe *datamodel.Recipe) *pipelinePB.Recipe {
 
 func marshalPipeline(pipeline *datamodel.Pipeline) *pipelinePB.Pipeline {
 	ret := &pipelinePB.Pipeline{
-		Id:          pipeline.Id,
+		Id:          uint64(pipeline.ID),
 		Name:        pipeline.Name,
 		Description: pipeline.Description,
-		Active:      pipeline.Active,
+		Status:      pipelinePB.Pipeline_Status(pipelinePB.Pipeline_Status_value[string(pipeline.Status)]),
 		CreatedAt:   timestamppb.New(pipeline.CreatedAt),
 		UpdatedAt:   timestamppb.New(pipeline.UpdatedAt),
 		FullName:    pipeline.FullName,