stub second pipeline

config.go

@@ -13,25 +13,26 @@ import (
 )
 
 type Config struct {
-	Port                   int
+	Port                         int
-	Debug                  bool
+	Debug                        bool
-	InitializeSlack        bool
+	InitializeSlack              bool
-	SlackToken             string
+	SlackToken                   string
-	SlackChannels          []string
+	SlackChannels                []string
-	DriverConn             string
+	DriverConn                   string
-	BasicAuthUser          string
+	BasicAuthUser                string
-	BasicAuthPassword      string
+	BasicAuthPassword            string
-	FillWithTestdata       bool
+	FillWithTestdata             bool
-	OllamaURL              string
+	OllamaURL                    string
-	OllamaModel            string
+	OllamaModel                  string
-	LocalCheckpoint        string
+	LocalCheckpoint              string
-	LocalTokenizer         string
+	LocalTokenizer               string
-	AssetPattern           string
+	AssetPattern                 string
-	DatacenterPattern      string
+	DatacenterPattern            string
-	EventNamePattern       string
+	EventNamePattern             string
-	driver                 Driver
+	driver                       Driver
-	ai                     AI
+	ai                           AI
-	slackToMessagePipeline Pipeline
+	slackToMessagePipeline       Pipeline
+	messageToPersistencePipeline Pipeline
 }
 
 var (
@@ -131,5 +132,11 @@ func newConfigFromEnv(ctx context.Context, getEnv func(string) string) (Config,
 	}
 	result.slackToMessagePipeline = slackToMessagePipeline
 
+	messageToPersistencePipeline, err := NewMessageToPersistencePipeline(ctx, result)
+	if err != nil {
+		return Config{}, err
+	}
+	result.messageToPersistencePipeline = messageToPersistencePipeline
+
 	return result, nil
 }

main.go

@@ -36,7 +36,7 @@ func run(ctx context.Context, cfg Config) error {
 	select {
 	case <-ctx.Done():
 		return ctx.Err()
-	case err := <-processPipelines(ctx, cfg.slackToMessagePipeline):
+	case err := <-processPipelines(ctx, cfg.slackToMessagePipeline, cfg.messageToPersistencePipeline):
 		return err
 	case err := <-listenAndServe(ctx, cfg):
 		return err
@@ -51,7 +51,7 @@ func processPipelines(ctx context.Context, first Pipeline, pipelines ...Pipeline
 	errs := make(chan error)
 	for i := range pipelines {
 		go func(i int) {
-			defer close(errs)
+			defer can()
 			select {
 			case errs <- pipelines[i].Process(ctx):
 			case <-ctx.Done():

persistance.go

@@ -0,0 +1,36 @@
+package main
+
+import (
+	"context"
+	"fmt"
+)
+
+type MessageToPersistence struct {
+	pipeline Pipeline
+}
+
+func NewMessageToPersistencePipeline(ctx context.Context, cfg Config) (Pipeline, error) {
+	reader, err := NewQueue(ctx, "fromSlack", cfg.driver)
+	if err != nil {
+		return Pipeline{}, err
+	}
+	writer, err := NewQueue(ctx, "fromMessage", cfg.driver)
+	if err != nil {
+		return Pipeline{}, err
+	}
+	return Pipeline{
+		writer:  writer,
+		reader:  reader,
+		process: newMessageToPersistenceProcess(cfg),
+	}, nil
+}
+
+func newMessageToPersistenceProcess(cfg Config) processFunc {
+	return func(ctx context.Context, slack []byte) ([]byte, error) {
+		m, err := ParseSlack(slack, cfg.AssetPattern, cfg.DatacenterPattern, cfg.EventNamePattern)
+		if err != nil {
+			return nil, fmt.Errorf("failed to deserialize slack %w: %s", err, slack)
+		}
+		return m.Serialize(), nil
+	}
+}