at least main_test runs

config.go

@@ -31,6 +31,7 @@ type Config struct {
 	EventNamePattern       string
 	driver                 Driver
 	ai                     AI
+	slackToMessagePipeline Pipeline
 }
 
 var (
@@ -124,5 +125,11 @@ func newConfigFromEnv(ctx context.Context, getEnv func(string) string) (Config,
 		result.ai = NewAINoop()
 	}
 
+	slackToMessagePipeline, err := NewSlackToMessagePipeline(ctx, result)
+	if err != nil {
+		return Config{}, err
+	}
+	result.slackToMessagePipeline = slackToMessagePipeline
+
 	return result, nil
 }

main.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"io"
 	"log"
@@ -49,13 +48,7 @@ func processSlackToMessagePipeline(ctx context.Context, cfg Config) chan error {
 	go func() {
 		defer close(errs)
 		select {
-		case errs <- func() error {
+		case errs <- cfg.slackToMessagePipeline.Process(ctx):
-			slackToMessagePipeline, err := NewSlackToMessagePipeline(ctx, cfg)
-			if err != nil {
-				return err
-			}
-			return slackToMessagePipeline.Process(ctx)
-		}():
 		case <-ctx.Done():
 		}
 	}()
@@ -139,14 +132,8 @@ func newHandlerPutAPIV1RPCScrapeSlack(cfg Config) http.HandlerFunc {
 		}
 		errs := []error{}
 		for _, messageJSON := range page.Messages {
-			m, err := ParseSlack(messageJSON, cfg.AssetPattern, cfg.DatacenterPattern, cfg.EventNamePattern)
+			if err := cfg.slackToMessagePipeline.reader.Enqueue(r.Context(), messageJSON); err != nil {
-			if errors.Is(err, ErrIrrelevantMessage) {
-			} else if err != nil {
 				errs = append(errs, err)
-			} else if err := cfg.storage.Upsert(r.Context(), m); err != nil {
-				errs = append(errs, err)
-			} else {
-				log.Printf("re-ingested %v", m.ID)
 			}
 		}
 
@@ -164,19 +151,7 @@ func newHandlerGetAPIV1EventNames(cfg Config) http.HandlerFunc {
 			return
 		}
 
-		since, err := parseSince(r.URL.Query().Get("since"))
+		http.Error(w, "not impl", http.StatusNotImplemented)
-		if err != nil {
-			http.Error(w, err.Error(), http.StatusBadRequest)
-			return
-		}
-
-		eventNames, err := cfg.storage.EventNamesSince(r.Context(), since)
-		if err != nil {
-			http.Error(w, err.Error(), http.StatusInternalServerError)
-			return
-		}
-
-		encodeResponse(w, r, map[string]any{"eventNames": eventNames})
 	}
 }
 
@@ -186,19 +161,7 @@ func newHandlerGetAPIV1Events(cfg Config) http.HandlerFunc {
 			return
 		}
 
-		since, err := parseSince(r.URL.Query().Get("since"))
+		http.Error(w, "not impl", http.StatusNotImplemented)
-		if err != nil {
-			http.Error(w, err.Error(), http.StatusBadRequest)
-			return
-		}
-
-		events, err := cfg.storage.EventsSince(r.Context(), since)
-		if err != nil {
-			http.Error(w, err.Error(), http.StatusInternalServerError)
-			return
-		}
-
-		encodeResponse(w, r, map[string]any{"events": events})
 	}
 }
 
@@ -208,19 +171,7 @@ func newHandlerGetAPIV1Messages(cfg Config) http.HandlerFunc {
 			return
 		}
 
-		since, err := parseSince(r.URL.Query().Get("since"))
+		http.Error(w, "not impl", http.StatusNotImplemented)
-		if err != nil {
-			http.Error(w, err.Error(), http.StatusBadRequest)
-			return
-		}
-
-		messages, err := cfg.storage.MessagesSince(r.Context(), since)
-		if err != nil {
-			http.Error(w, err.Error(), http.StatusInternalServerError)
-			return
-		}
-
-		encodeResponse(w, r, map[string]any{"messages": messages})
 	}
 }
 
@@ -230,19 +181,7 @@ func newHandlerGetAPIV1Threads(cfg Config) http.HandlerFunc {
 			return
 		}
 
-		since, err := parseSince(r.URL.Query().Get("since"))
+		http.Error(w, "not impl", http.StatusNotImplemented)
-		if err != nil {
-			http.Error(w, err.Error(), http.StatusBadRequest)
-			return
-		}
-
-		threads, err := cfg.storage.ThreadsSince(r.Context(), since)
-		if err != nil {
-			http.Error(w, err.Error(), http.StatusInternalServerError)
-			return
-		}
-
-		encodeResponse(w, r, map[string]any{"threads": threads})
 	}
 }
 
@@ -253,14 +192,9 @@ func newHandlerGetAPIV1ThreadsThread(cfg Config) http.HandlerFunc {
 		}
 
 		thread := strings.Split(strings.Split(r.URL.Path, "/threads/")[1], "/")[0]
+		_ = thread
 
-		messages, err := cfg.storage.Thread(r.Context(), thread)
+		http.Error(w, "not impl", http.StatusNotImplemented)
-		if err != nil {
-			http.Error(w, err.Error(), http.StatusInternalServerError)
-			return
-		}
-
-		encodeResponse(w, r, map[string]any{"thread": messages})
 	}
 }
 
@@ -322,20 +256,12 @@ func _newHandlerPostAPIV1EventsSlack(cfg Config) http.HandlerFunc {
 			return
 		}
 
-		m, err := ParseSlack(b, cfg.AssetPattern, cfg.DatacenterPattern, cfg.EventNamePattern)
+		if err := cfg.slackToMessagePipeline.reader.Enqueue(r.Context(), b); err != nil {
-		if errors.Is(err, ErrIrrelevantMessage) {
+			log.Printf("failed to ingest: %v", err)
-			return
-		} else if err != nil {
-			http.Error(w, err.Error(), http.StatusBadRequest)
-			return
-		}
-
-		if err := cfg.storage.Upsert(r.Context(), m); err != nil {
-			log.Printf("failed to ingest %+v: %v", m, err)
 			http.Error(w, err.Error(), http.StatusInternalServerError)
 			return
 		}
-		log.Printf("ingested %v", m.ID)
+		log.Printf("ingested")
 	}
 }
 

main_test.go

@@ -43,9 +43,8 @@ func TestRun(t *testing.T) {
 	cfg.AssetPattern = renderAssetPattern
 	cfg.EventNamePattern = renderEventNamePattern
 	cfg.Port = port
-	cfg.driver = NewRAM()
+	cfg.driver, _ = NewDriver(ctx, "")
-	cfg.storage = NewStorage(cfg.driver)
+	cfg.slackToMessagePipeline, _ = NewSlackToMessagePipeline(ctx, cfg)
-	cfg.queue = NewQueue(cfg.driver)
 	cfg.SlackToken = "redacted"
 	cfg.SlackChannels = []string{"C06U1DDBBU4"}