package main import ( "bytes" "context" "encoding/csv" "encoding/json" "errors" "fmt" "io" "log" "net" "net/http" "os/signal" "strconv" "strings" "syscall" "time" ) func main() { ctx, can := signal.NotifyContext(context.Background(), syscall.SIGINT) defer can() cfg, err := newConfig(ctx) if err != nil { panic(err) } defer cfg.driver.Close() if err := run(ctx, cfg); err != nil && ctx.Err() == nil { panic(err) } } func run(ctx context.Context, cfg Config) error { select { case <-ctx.Done(): return ctx.Err() case err := <-listenAndServe(ctx, cfg): return err } } func listenAndServe(ctx context.Context, cfg Config) chan error { s := http.Server{ Addr: fmt.Sprintf(":%d", cfg.Port), Handler: http.HandlerFunc(newHandler(cfg)), BaseContext: func(net.Listener) context.Context { return ctx }, } errc := make(chan error) go func() { defer close(errc) log.Printf("listening on %s", s.Addr) errc <- s.ListenAndServe() }() return errc } func newHandler(cfg Config) http.HandlerFunc { mux := http.NewServeMux() mux.Handle("POST /api/v1/events/slack", http.HandlerFunc(newHandlerPostAPIV1EventsSlack(cfg))) mux.Handle("GET /api/v1/messages", http.HandlerFunc(newHandlerGetAPIV1Messages(cfg))) mux.Handle("GET /api/v1/threads", http.HandlerFunc(newHandlerGetAPIV1Threads(cfg))) mux.Handle("GET /api/v1/threads/{thread}", http.HandlerFunc(newHandlerGetAPIV1ThreadsThread(cfg))) return func(w http.ResponseWriter, r *http.Request) { if cfg.Debug { b, _ := io.ReadAll(r.Body) r.Body = io.NopCloser(bytes.NewReader(b)) log.Printf("%s %s | %s", r.Method, r.URL, b) } mux.ServeHTTP(w, r) } } func newHandlerGetAPIV1Messages(cfg Config) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { if !basicAuth(cfg, w, r) { return } since, err := parseSince(r.URL.Query().Get("since")) 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 } writeJSON(w, map[string]any{"messages": messages}) } } func newHandlerGetAPIV1Threads(cfg Config) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { if !basicAuth(cfg, w, r) { return } since, err := parseSince(r.URL.Query().Get("since")) 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 } writeJSON(w, map[string]any{"threads": threads}) } } func newHandlerGetAPIV1ThreadsThread(cfg Config) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { if !basicAuth(cfg, w, r) { return } thread := strings.Split(strings.Split(r.URL.Path, "/threads/")[1], "/")[0] messages, err := cfg.storage.Thread(r.Context(), thread) if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } writeJSON(w, map[string]any{"thread": map[string]any{"messages": messages}}) } } func basicAuth(cfg Config, w http.ResponseWriter, r *http.Request) bool { if u, p, _ := r.BasicAuth(); u != cfg.BasicAuthUser || p != cfg.BasicAuthPassword { http.Error(w, "shoo", http.StatusForbidden) return false } return true } func newHandlerPostAPIV1EventsSlack(cfg Config) http.HandlerFunc { if cfg.InitializeSlack { return handlerPostAPIV1EventsSlackInitialize } return _newHandlerPostAPIV1EventsSlack(cfg) } func handlerPostAPIV1EventsSlackInitialize(w http.ResponseWriter, r *http.Request) { b, _ := io.ReadAll(r.Body) var challenge struct { Token string Challenge string Type string } if err := json.Unmarshal(b, &challenge); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } writeJSON(w, map[string]any{"challenge": challenge.Challenge}) } func _newHandlerPostAPIV1EventsSlack(cfg Config) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { b, _ := io.ReadAll(r.Body) r.Body = io.NopCloser(bytes.NewReader(b)) var allowList struct { Token string Event struct { Channel string } } if err := json.Unmarshal(b, &allowList); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } else if allowList.Token != cfg.SlackToken { http.Error(w, "invalid .token", http.StatusForbidden) return } else if !func() bool { for _, slackChannel := range cfg.SlackChannels { if slackChannel == allowList.Event.Channel { return true } } return false }() { return } m, err := ParseSlack(b) if errors.Is(err, ErrIrrelevantMessage) { 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) } } func parseSince(s string) (time.Time, error) { if s == "" { return time.Unix(0, 0), nil } if n, err := strconv.ParseInt(s, 10, 64); err != nil { } else { return time.Unix(n, 0), nil } if t, err := time.Parse(time.RFC3339, s); err != nil { } else { return t, nil } if t, err := time.Parse(time.RFC3339Nano, s); err != nil { } else { return t, nil } if t, err := time.ParseInLocation(time.DateOnly, s, time.Local); err != nil { } else { return t, nil } return time.Time{}, fmt.Errorf("failed to parse since=%q", s) } func writeJSON(w http.ResponseWriter, v interface{}) error { return json.NewEncoder(w).Encode(v) } func writeCSV(w http.ResponseWriter, fields []string, values [][]string) error { enc := csv.NewWriter(w) if err := enc.Write(fields); err != nil { return err } return enc.WriteAll(values) }