feat: add more fields to eventbus and subscription (#316)

Author: wenfengwang

internal/controller/eventbus/controller.go

@@ -133,10 +133,13 @@ func (ctrl *controller) CreateEventBus(ctx context.Context,
 		return nil, err
 	}
 	eb := &metadata.Eventbus{
-		ID:        id,
-		Name:      req.Name,
-		LogNumber: int(logNum),
-		EventLogs: make([]*metadata.Eventlog, int(logNum)),
+		ID:          id,
+		Name:        req.Name,
+		LogNumber:   int(logNum),
+		EventLogs:   make([]*metadata.Eventlog, int(logNum)),
+		Description: req.Description,
+		CreatedAt:   time.Now(),
+		UpdatedAt:   time.Now(),
 	}
 	exist, err := ctrl.kvStore.Exists(ctx, metadata.GetEventbusMetadataKey(eb.Name))
 	if err != nil {
@@ -204,8 +207,6 @@ func (ctrl *controller) getEventbus(name string) (*metapb.EventBus, error) {
 	}
 
 	ebMD := metadata.Convert2ProtoEventBus(_eb)[0]
-	ebMD.Name = _eb.Name
-	ebMD.Logs = metadata.Convert2ProtoEventLog(_eb.EventLogs...)
 	addrs := make([]string, 0)
 	for _, v := range ctrl.cfg.Topology {
 		addrs = append(addrs, v)

internal/controller/eventbus/controller_test.go

@@ -65,7 +65,7 @@ func TestController_CreateEventBus(t *testing.T) {
 			})
 			So(err, ShouldBeNil)
 			So(res.Name, ShouldEqual, "test-1")
-			So(res.Id, ShouldEqual, el.EventbusID.Uint64())
+			So(res.Id, ShouldNotEqual, 0)
 			So(res.Logs, ShouldHaveLength, 1)
 			So(res.LogNumber, ShouldEqual, 1)
 			So(res.Logs[0].EventBusName, ShouldEqual, "test-1")

internal/controller/eventbus/metadata/info.go

@@ -16,25 +16,34 @@ package metadata
 
 import (
 	"encoding/json"
+	"time"
 
 	"github.com/linkall-labs/vanus/internal/primitive/vanus"
 	"github.com/linkall-labs/vanus/proto/pkg/meta"
 )
 
 type Eventbus struct {
-	ID        vanus.ID    `json:"id"`
-	Name      string      `json:"name"`
-	LogNumber int         `json:"log_number"`
-	EventLogs []*Eventlog `json:"event_logs"`
+	ID          vanus.ID    `json:"id"`
+	Name        string      `json:"name"`
+	LogNumber   int         `json:"log_number"`
+	EventLogs   []*Eventlog `json:"event_logs"`
+	Description string      `json:"description"`
+	CreatedAt   time.Time   `json:"created_at"`
+	UpdatedAt   time.Time   `json:"updated_at"`
 }
 
 func Convert2ProtoEventBus(ins ...*Eventbus) []*meta.EventBus {
 	pebs := make([]*meta.EventBus, len(ins))
 	for idx := 0; idx < len(ins); idx++ {
 		eb := ins[idx]
 		pebs[idx] = &meta.EventBus{
-			Name:      eb.Name,
-			LogNumber: int32(eb.LogNumber),
+			Name:        eb.Name,
+			LogNumber:   int32(eb.LogNumber),
+			Logs:        Convert2ProtoEventLog(eb.EventLogs...),
+			Id:          eb.ID.Uint64(),
+			Description: eb.Description,
+			CreatedAt:   eb.CreatedAt.UnixMilli(),
+			UpdatedAt:   eb.UpdatedAt.UnixMilli(),
 		}
 	}
 	return pebs

internal/controller/trigger/controller.go

@@ -141,6 +141,8 @@ func (ctrl *controller) CreateSubscription(ctx context.Context,
 	}
 	sub := convert.FromPbSubscriptionRequest(request.Subscription)
 	sub.ID, err = vanus.NewID()
+	sub.CreatedAt = time.Now()
+	sub.UpdatedAt = time.Now()
 	if err != nil {
 		return nil, err
 	}
@@ -186,6 +188,7 @@ func (ctrl *controller) UpdateSubscription(ctx context.Context,
 	if !change {
 		return nil, errors.ErrInvalidRequest.WithMessage("no change")
 	}
+	sub.UpdatedAt = time.Now()
 	sub.Phase = metadata.SubscriptionPhasePending
 	if err := ctrl.subscriptionManager.UpdateSubscription(ctx, sub); err != nil {
 		return nil, err

internal/controller/trigger/metadata/info.go

@@ -75,6 +75,11 @@ type Subscription struct {
 	ProtocolSetting    *primitive.ProtocolSetting      `json:"protocol_settings,omitempty"`
 	EventBus           string                          `json:"eventbus"`
 	Transformer        *primitive.Transformer          `json:"transformer,omitempty"`
+	Name               string                          `json:"name"`
+	Disable            bool                            `json:"disable"`
+	Description        string                          `json:"description"`
+	CreatedAt          time.Time                       `json:"created_at"`
+	UpdatedAt          time.Time                       `json:"updated_at"`
 
 	// not from api
 	Phase         SubscriptionPhase `json:"phase"`
@@ -89,6 +94,14 @@ func (s *Subscription) Update(update *Subscription) bool {
 		change = true
 		s.Source = update.Source
 	}
+	if s.Description != update.Description {
+		change = true
+		s.Description = update.Description
+	}
+	if s.Disable != update.Disable {
+		change = true
+		s.Disable = update.Disable
+	}
 	if !reflect.DeepEqual(s.Types, update.Types) {
 		change = true
 		s.Types = update.Types

internal/convert/convert.go

@@ -38,6 +38,9 @@ func FromPbSubscriptionRequest(sub *ctrl.SubscriptionRequest) *metadata.Subscrip
 		Filters:            fromPbFilters(sub.Filters),
 		Transformer:        fromPbTransformer(sub.Transformer),
 		EventBus:           sub.EventBus,
+		Name:               sub.Name,
+		Disable:            sub.Disable,
+		Description:        sub.Description,
 	}
 	return to
 }
@@ -282,6 +285,11 @@ func ToPbSubscription(sub *metadata.Subscription, offsets info.ListOffsetInfo) *
 		Filters:          toPbFilters(sub.Filters),
 		Transformer:      ToPbTransformer(sub.Transformer),
 		Offsets:          ToPbOffsetInfos(offsets),
+		Name:             sub.Name,
+		Disable:          sub.Disable,
+		Description:      sub.Description,
+		CreatedAt:        sub.CreatedAt.UnixMilli(),
+		UpdatedAt:        sub.UpdatedAt.UnixMilli(),
 	}
 	return to
 }

internal/primitive/vanus/id.go

@@ -16,6 +16,7 @@ package vanus
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"strconv"
 	"sync"
@@ -222,7 +223,14 @@ func NewIDFromUint64(id uint64) ID {
 	return ID(id)
 }
 
+var (
+	ErrEmptyID = errors.New("id: empty")
+)
+
 func NewIDFromString(id string) (ID, error) {
+	if id == "" {
+		return emptyID, ErrEmptyID
+	}
 	i, err := strconv.ParseUint(id, base, bitSize)
 	if err != nil {
 		return emptyID, err