add event recording diag command

cmd/slackdump/internal/cfg/cfg.go

@@ -37,14 +37,14 @@ const (
 	OmitAuthFlags FlagMask = 1 << iota
 	OmitDownloadFlag
 	OmitConfigFlag
-	OmitBaseLoc
+	OmitBaseLocFlag
 	OmitCacheDir
 	OmitWorkspaceFlag
 	OmitUserCacheFlag
 
 	OmitAll = OmitConfigFlag |
 		OmitDownloadFlag |
-		OmitBaseLoc |
+		OmitBaseLocFlag |
 		OmitCacheDir |
 		OmitWorkspaceFlag |
 		OmitAuthFlags |
@@ -70,7 +70,7 @@ func SetBaseFlags(fs *flag.FlagSet, mask FlagMask) {
 	if mask&OmitConfigFlag == 0 {
 		fs.StringVar(&ConfigFile, "api-config", "", "configuration `file` with Slack API limits overrides.\nYou can generate one with default values with 'slackdump config new`")
 	}
-	if mask&OmitBaseLoc == 0 {
+	if mask&OmitBaseLocFlag == 0 {
 		base := fmt.Sprintf("slackdump_%s.zip", time.Now().Format("20060102_150405"))
 		fs.StringVar(&SlackConfig.BaseLocation, "base", osenv.Value("BASE_LOC", base), "a `location` (a directory or a ZIP file) on the local disk to save\ndownloaded files to.")
 	}

cmd/slackdump/internal/diag/diag.go

@@ -23,5 +23,6 @@ open an issue on Github.
 		CmdEzTest,
 		CmdThread,
 		CmdObfuscate,
+		CmdRecord,
 	},
 }

cmd/slackdump/internal/diag/record.go

@@ -0,0 +1,66 @@
+package diag
+
+import (
+	"context"
+	"errors"
+	"io"
+	"os"
+	"time"
+
+	"github.com/rusq/slackdump/v2"
+	"github.com/rusq/slackdump/v2/auth"
+	"github.com/rusq/slackdump/v2/cmd/slackdump/internal/cfg"
+	"github.com/rusq/slackdump/v2/cmd/slackdump/internal/golang/base"
+	"github.com/rusq/slackdump/v2/internal/processors"
+)
+
+var CmdRecord = &base.Command{
+	UsageLine:   "slackdump diag record [options] <channel>",
+	Short:       "dump slack data in a event record format",
+	FlagMask:    cfg.OmitBaseLocFlag | cfg.OmitDownloadFlag,
+	PrintFlags:  true,
+	RequireAuth: true,
+}
+
+func init() {
+	// break init cycle
+	CmdRecord.Run = runRecord
+}
+
+var output = CmdRecord.Flag.String("output", "", "output file")
+
+func runRecord(ctx context.Context, cmd *base.Command, args []string) error {
+	if len(args) == 0 {
+		return errors.New("missing channel argument")
+	}
+
+	prov, err := auth.FromContext(ctx)
+	if err != nil {
+		return err
+	}
+	sess, err := slackdump.New(ctx, prov, cfg.SlackConfig)
+	if err != nil {
+		return err
+	}
+	defer sess.Close()
+
+	var w io.Writer
+	if *output == "" {
+		w = os.Stdout
+	} else {
+		f, err := os.Create("output.jsonl")
+		if err != nil {
+			return err
+		}
+		defer f.Close()
+		w = f
+	}
+
+	rec := processors.NewRecorder(w)
+	defer rec.Close()
+
+	if err := sess.Stream(ctx, args[0], rec, time.Time{}, time.Time{}); err != nil {
+		return err
+	}
+	return nil
+}

internal/fixtures/events.go

@@ -0,0 +1,25 @@
+package fixtures
+
+import (
+	"bytes"
+	"compress/gzip"
+	_ "embed"
+	"io"
+)
+
+// To generate the events.jsonl.gz file:
+//   1. Record events from slackdump
+//   2. Obfuscate them with ./cmd/slackdump diag obfuscate -i clear.jsonl -o events.jsonl
+//   3. Compress them with gzip -9 -c events.jsonl > events.jsonl.gz
+
+//go:embed assets/events.jsonl.gz
+var eventsJsonlGz []byte
+
+// EventsJSONL returns a reader for the events.jsonl.gz file.
+func EventsJSONL() io.ReadCloser {
+	gz, err := gzip.NewReader(bytes.NewReader(eventsJsonlGz))
+	if err != nil {
+		panic(err)
+	}
+	return gz
+}

internal/processors/recorder.go

@@ -54,9 +54,9 @@ func (e *Event) ID() string {
 	return "<empty>"
 }
 
-func NewRecorder(wc io.Writer) *Recorder {
+func NewRecorder(w io.Writer) *Recorder {
 	rec := &Recorder{
-		w:      wc,
+		w:      w,
 		events: make(chan Event),
 		resp:   make(chan error, 1),
 	}

slackdump.go

@@ -8,6 +8,7 @@ import (
 	"log"
 	"os"
 	"runtime/trace"
+	"time"
 
 	"github.com/go-playground/validator/v10"
 	"github.com/slack-go/slack"
@@ -18,6 +19,7 @@ import (
 	"github.com/rusq/slackdump/v2/auth"
 	"github.com/rusq/slackdump/v2/fsadapter"
 	"github.com/rusq/slackdump/v2/internal/network"
+	"github.com/rusq/slackdump/v2/internal/processors"
 	"github.com/rusq/slackdump/v2/logger"
 	"github.com/rusq/slackdump/v2/types"
 )
@@ -257,3 +259,12 @@ func (s *Session) propagateLogger() {
 func (s *Session) Info() *WorkspaceInfo {
 	return s.wspInfo
 }
+
+// Stream streams the channel, calling Channeler functions for each chunk.
+func (s *Session) Stream(ctx context.Context, link string, proc processors.Channeler, oldest, latest time.Time) error {
+	ctx, task := trace.NewTask(ctx, "Stream")
+	defer task.End()
+
+	cs := newChannelStream(s.client, &s.cfg.Limits, oldest, latest)
+	return cs.Stream(ctx, link, proc)
+}