Improve bus (#3)

* Improve bus * Add ticket log
2026-01-24 06:53:27 +01:00 · 2021-12-27 19:08:07 +01:00
parent 1fade14ba5
commit b5dd0cfacd
50 changed files with 756 additions and 456 deletions
--- a/bus/bus.go
+++ b/bus/bus.go
@@ -4,16 +4,7 @@ import (
 	"encoding/json"
 	"log"

-	"github.com/arangodb/go-driver"
 	emitter "github.com/emitter-io/go/v2"
-
-	"github.com/SecurityBrewery/catalyst/generated/models"
-)
-
-const (
-	channelUpdate = "data"
-	channelJob    = "job"
-	channelResult = "result"
 )

 type Bus struct {
@@ -22,25 +13,13 @@ type Bus struct {
 }

 type Config struct {
-	Host         string
-	Key          string
-	resultBusKey string
-	jobBusKey    string
-	dataBusKey   string
-	APIUrl       string
-}
-
-type JobMsg struct {
-	ID         string          `json:"id"`
-	Automation string          `json:"automation"`
-	Origin     *models.Origin  `json:"origin"`
-	Message    *models.Message `json:"message"`
-}
-
-type ResultMsg struct {
-	Automation string                 `json:"automation"`
-	Data       map[string]interface{} `json:"data,omitempty"`
-	Target     *models.Origin         `json:"target"`
+	Host                 string
+	Key                  string
+	databaseUpdateBusKey string
+	jobBusKey            string
+	resultBusKey         string
+	requestKey           string
+	APIUrl               string
 }

 func New(c *Config) (*Bus, error) {
@@ -51,7 +30,7 @@ func New(c *Config) (*Bus, error) {
 		return nil, err
 	}

-	c.dataBusKey, err = client.GenerateKey(c.Key, channelUpdate+"/", "rwls", 0)
+	c.databaseUpdateBusKey, err = client.GenerateKey(c.Key, channelDatabaseUpdate+"/", "rwls", 0)
 	if err != nil {
 		return nil, err
 	}
@@ -63,30 +42,14 @@ func New(c *Config) (*Bus, error) {
 	if err != nil {
 		return nil, err
 	}
+	c.requestKey, err = client.GenerateKey(c.Key, ChannelRequest+"/", "rwls", 0)
+	if err != nil {
+		return nil, err
+	}

 	return &Bus{config: c, client: client}, err
 }

-func (b *Bus) PublishUpdate(ids []driver.DocumentID) error {
-	return b.jsonPublish(ids, channelUpdate, b.config.dataBusKey)
-}
-
-func (b *Bus) PublishJob(id, automation string, payload interface{}, context *models.Context, origin *models.Origin) error {
-	return b.jsonPublish(&JobMsg{
-		ID:         id,
-		Automation: automation,
-		Origin:     origin,
-		Message: &models.Message{
-			Context: context,
-			Payload: payload,
-		},
-	}, channelJob, b.config.jobBusKey)
-}
-
-func (b *Bus) PublishResult(automation string, data map[string]interface{}, target *models.Origin) error {
-	return b.jsonPublish(&ResultMsg{Automation: automation, Data: data, Target: target}, channelResult, b.config.resultBusKey)
-}
-
 func (b *Bus) jsonPublish(msg interface{}, channel, key string) error {
 	payload, err := json.Marshal(msg)
 	if err != nil {
@@ -96,39 +59,6 @@ func (b *Bus) jsonPublish(msg interface{}, channel, key string) error {
 	return b.client.Publish(key, channel, payload)
 }

-func (b *Bus) SubscribeUpdate(f func(ids []driver.DocumentID)) error {
-	return b.safeSubscribe(b.config.dataBusKey, channelUpdate, func(c *emitter.Client, m emitter.Message) {
-		var msg []driver.DocumentID
-		if err := json.Unmarshal(m.Payload(), &msg); err != nil {
-			log.Println(err)
-			return
-		}
-		go f(msg)
-	})
-}
-
-func (b *Bus) SubscribeJob(f func(msg *JobMsg)) error {
-	return b.safeSubscribe(b.config.jobBusKey, channelJob, func(c *emitter.Client, m emitter.Message) {
-		var msg JobMsg
-		if err := json.Unmarshal(m.Payload(), &msg); err != nil {
-			log.Println(err)
-			return
-		}
-		go f(&msg)
-	})
-}
-
-func (b *Bus) SubscribeResult(f func(msg *ResultMsg)) error {
-	return b.safeSubscribe(b.config.resultBusKey, channelResult, func(c *emitter.Client, m emitter.Message) {
-		var msg ResultMsg
-		if err := json.Unmarshal(m.Payload(), &msg); err != nil {
-			log.Println(err)
-			return
-		}
-		go f(&msg)
-	})
-}
-
 func (b *Bus) safeSubscribe(key, channel string, handler func(c *emitter.Client, m emitter.Message)) error {
 	defer func() {
 		if r := recover(); r != nil {
--- a/bus/databaseupdate.go
+++ b/bus/databaseupdate.go
@@ -0,0 +1,42 @@
+package bus
+
+import (
+	"encoding/json"
+	"log"
+
+	"github.com/arangodb/go-driver"
+	emitter "github.com/emitter-io/go/v2"
+)
+
+const channelDatabaseUpdate = "databaseupdate"
+
+type DatabaseUpdateType string
+
+const (
+	DatabaseEntryRead    DatabaseUpdateType = "read"
+	DatabaseEntryCreated DatabaseUpdateType = "created"
+	DatabaseEntryUpdated DatabaseUpdateType = "updated"
+)
+
+type DatabaseUpdateMsg struct {
+	IDs  []driver.DocumentID `json:"ids"`
+	Type DatabaseUpdateType  `json:"type"`
+}
+
+func (b *Bus) PublishDatabaseUpdate(ids []driver.DocumentID, databaseUpdateType DatabaseUpdateType) error {
+	return b.jsonPublish(&DatabaseUpdateMsg{
+		IDs:  ids,
+		Type: databaseUpdateType,
+	}, channelDatabaseUpdate, b.config.databaseUpdateBusKey)
+}
+
+func (b *Bus) SubscribeDatabaseUpdate(f func(msg *DatabaseUpdateMsg)) error {
+	return b.safeSubscribe(b.config.databaseUpdateBusKey, channelDatabaseUpdate, func(c *emitter.Client, m emitter.Message) {
+		var msg DatabaseUpdateMsg
+		if err := json.Unmarshal(m.Payload(), &msg); err != nil {
+			log.Println(err)
+			return
+		}
+		go f(&msg)
+	})
+}
--- a/bus/job.go
+++ b/bus/job.go
@@ -0,0 +1,42 @@
+package bus
+
+import (
+	"encoding/json"
+	"log"
+
+	emitter "github.com/emitter-io/go/v2"
+
+	"github.com/SecurityBrewery/catalyst/generated/models"
+)
+
+const channelJob = "job"
+
+type JobMsg struct {
+	ID         string          `json:"id"`
+	Automation string          `json:"automation"`
+	Origin     *models.Origin  `json:"origin"`
+	Message    *models.Message `json:"message"`
+}
+
+func (b *Bus) PublishJob(id, automation string, payload interface{}, context *models.Context, origin *models.Origin) error {
+	return b.jsonPublish(&JobMsg{
+		ID:         id,
+		Automation: automation,
+		Origin:     origin,
+		Message: &models.Message{
+			Context: context,
+			Payload: payload,
+		},
+	}, channelJob, b.config.jobBusKey)
+}
+
+func (b *Bus) SubscribeJob(f func(msg *JobMsg)) error {
+	return b.safeSubscribe(b.config.jobBusKey, channelJob, func(c *emitter.Client, m emitter.Message) {
+		var msg JobMsg
+		if err := json.Unmarshal(m.Payload(), &msg); err != nil {
+			log.Println(err)
+			return
+		}
+		go f(&msg)
+	})
+}
--- a/bus/request.go
+++ b/bus/request.go
@@ -0,0 +1,36 @@
+package bus
+
+import (
+	"encoding/json"
+	"log"
+
+	"github.com/arangodb/go-driver"
+	emitter "github.com/emitter-io/go/v2"
+)
+
+const ChannelRequest = "request"
+
+type RequestMsg struct {
+	IDs      []driver.DocumentID `json:"ids"`
+	Function string              `json:"function"`
+	User     string              `json:"user"`
+}
+
+func (b *Bus) PublishRequest(user, f string, ids []driver.DocumentID) error {
+	return b.jsonPublish(&RequestMsg{
+		User:     user,
+		Function: f,
+		IDs:      ids,
+	}, ChannelRequest, b.config.requestKey)
+}
+
+func (b *Bus) SubscribeRequest(f func(msg *RequestMsg)) error {
+	return b.safeSubscribe(b.config.requestKey, ChannelRequest, func(c *emitter.Client, m emitter.Message) {
+		var msg RequestMsg
+		if err := json.Unmarshal(m.Payload(), &msg); err != nil {
+			log.Println(err)
+			return
+		}
+		go f(&msg)
+	})
+}
--- a/bus/result.go
+++ b/bus/result.go
@@ -0,0 +1,33 @@
+package bus
+
+import (
+	"encoding/json"
+	"log"
+
+	emitter "github.com/emitter-io/go/v2"
+
+	"github.com/SecurityBrewery/catalyst/generated/models"
+)
+
+const channelResult = "result"
+
+type ResultMsg struct {
+	Automation string                 `json:"automation"`
+	Data       map[string]interface{} `json:"data,omitempty"`
+	Target     *models.Origin         `json:"target"`
+}
+
+func (b *Bus) PublishResult(automation string, data map[string]interface{}, target *models.Origin) error {
+	return b.jsonPublish(&ResultMsg{Automation: automation, Data: data, Target: target}, channelResult, b.config.resultBusKey)
+}
+
+func (b *Bus) SubscribeResult(f func(msg *ResultMsg)) error {
+	return b.safeSubscribe(b.config.resultBusKey, channelResult, func(c *emitter.Client, m emitter.Message) {
+		var msg ResultMsg
+		if err := json.Unmarshal(m.Payload(), &msg); err != nil {
+			log.Println(err)
+			return
+		}
+		go f(&msg)
+	})
+}