adding search controller, directory processor and command

cmd/slackdump/internal/record/search.go

@@ -0,0 +1,46 @@
+package record
+
+import (
+	"context"
+	"errors"
+	"strings"
+
+	"github.com/rusq/slackdump/v3/cmd/slackdump/internal/cfg"
+	"github.com/rusq/slackdump/v3/cmd/slackdump/internal/golang/base"
+	"github.com/rusq/slackdump/v3/internal/chunk"
+	"github.com/rusq/slackdump/v3/internal/chunk/control"
+)
+
+var CmdSearch = &base.Command{
+	UsageLine:   "slackdump search [flags] query terms",
+	Short:       "records search results matching the given query",
+	Long:        `Searches for messages matching criteria.`,
+	RequireAuth: true,
+	Run:         runSearch,
+	PrintFlags:  true,
+}
+
+func runSearch(ctx context.Context, cmd *base.Command, args []string) error {
+	if len(args) == 0 {
+		base.SetExitStatus(base.SInvalidParameters)
+		return errors.New("missing query parameter")
+	}
+	query := strings.Join(args, " ")
+
+	sess, err := cfg.SlackdumpSession(ctx)
+	if err != nil {
+		return err
+	}
+
+	cd, err := chunk.CreateDir(cfg.Output)
+	if err != nil {
+		base.SetExitStatus(base.SGenericError)
+		return err
+	}
+	defer cd.Close()
+
+	stream := sess.Stream()
+	ctrl := control.NewSearch(cd, stream)
+
+	return ctrl.Search(ctx, query)
+}

cmd/slackdump/main.go

@@ -44,6 +44,7 @@ func init() {
 		export.CmdExport,
 		dump.CmdDump,
 		record.CmdRecord,
+		record.CmdSearch,
 		convertcmd.CmdConvert,
 		list.CmdList,
 		emoji.CmdEmoji,

internal/chunk/control/interfaces.go

@@ -14,6 +14,7 @@ type Streamer interface {
 	ListChannels(ctx context.Context, proc processor.Channels, p *slack.GetConversationsParameters) error
 	Users(ctx context.Context, proc processor.Users, opt ...slack.GetUsersOption) error
 	WorkspaceInfo(ctx context.Context, proc processor.WorkspaceInfo) error
+	SearchMessages(ctx context.Context, proc processor.Search, query string) error
 }
 
 type TransformStarter interface {

internal/chunk/control/search.go

@@ -0,0 +1,37 @@
+package control
+
+import (
+	"context"
+	"time"
+
+	"github.com/rusq/slackdump/v3"
+	"github.com/rusq/slackdump/v3/internal/chunk"
+	"github.com/rusq/slackdump/v3/logger"
+	"golang.org/x/sync/errgroup"
+)
+
+type Search struct {
+	cd *chunk.Directory
+	s  *slackdump.Stream
+	lg logger.Interface
+}
+
+func NewSearch(cd *chunk.Directory, s *slackdump.Stream) *Search {
+	return &Search{cd: cd, s: s, lg: logger.Default}
+}
+
+func (s *Search) Search(ctx context.Context, query string) error {
+	var eg errgroup.Group
+	start := time.Now()
+	eg.Go(func() error {
+		return searchWorker(ctx, s.s, s.cd, query)
+	})
+	eg.Go(func() error {
+		return workspaceWorker(ctx, s.s, s.cd)
+	})
+	if err := eg.Wait(); err != nil {
+		return err
+	}
+	s.lg.Printf("search for query %q completed in: %s", query, time.Since(start))
+	return nil
+}

internal/chunk/control/workers.go

@@ -67,3 +67,18 @@ func workspaceWorker(ctx context.Context, s Streamer, cd *chunk.Directory) error
 	lg.Debug("workspaceWorker done")
 	return nil
 }
+
+func searchWorker(ctx context.Context, s Streamer, cd *chunk.Directory, query string) error {
+	lg := logger.FromContext(ctx)
+	lg.Debug("searchWorker started")
+	search, err := dirproc.NewSearch(cd)
+	if err != nil {
+		return err
+	}
+	defer search.Close()
+	if err := s.SearchMessages(ctx, search, query); err != nil {
+		return err
+	}
+	lg.Debug("searchWorker done")
+	return nil
+}

internal/chunk/dirproc/search.go

@@ -0,0 +1,15 @@
+package dirproc
+
+import "github.com/rusq/slackdump/v3/internal/chunk"
+
+type Search struct {
+	*baseproc
+}
+
+func NewSearch(dir *chunk.Directory) (*Search, error) {
+	p, err := newBaseProc(dir, "search")
+	if err != nil {
+		return nil, err
+	}
+	return &Search{baseproc: p}, nil
+}