WIP: database processor

.codespellrc

@@ -4,4 +4,4 @@ skip = .git,go.sum,.codespellrc,.goreleaser.yaml
 check-hidden = true
 # ignore-regex = 
 # ignore some variable names
-ignore-words-list = ser,ans,auther,Nd,FO,Nexted
+ignore-words-list = ser,ans,auther,Nd,FO,Nexted,flate,efore,aci,fter

.github/workflows/go.yml

@@ -16,7 +16,7 @@ jobs:
     - name: Set up Go
       uses: actions/setup-go@v5
       with:
-        go-version: "1.23"
+        go-version: "1.24"
 
     - name: Build
       run: go build -v ./...
@@ -36,7 +36,7 @@ jobs:
     - name: Set up Go
       uses: actions/setup-go@v5
       with:
-        go-version: "1.23"
+        go-version: "1.24"
 
     - name: Build
       run: go build -v ./...

.gitignore

@@ -66,3 +66,9 @@ debug.test*
 !internal/viewer/templates/*.html
 !internal/viewer/renderer/templates/*.html
 !internal/fixtures/**/*.json
+*.sqlite*
+*.db
+# profiler files
+*.pprof
+*.prof
+*.pf

Makefile

@@ -17,7 +17,7 @@ DISTFILES=README.md LICENSE
 ZIPFILES=$(foreach s,$(OSES),$(OUTPUT)-$s.zip)
 
 
-.PHONY: dist all test
+.PHONY: dist all
 
 # special guest.
 $(OUTPUT)-windows.zip: EXECUTABLE=$(OUTPUT).exe
@@ -57,9 +57,12 @@ arm_%:
 
 clean:
 	-rm slackdump slackdump.exe $(wildcard *.zip)
+	-rm -rf slackdump_$(shell date +%Y)*
+.PHONY: clean
 
 test:
 	go test -race -cover ./...
+.PHONY: test
 
 aurtest:
 	GOFLAGS="-buildmode=pie -trimpath -ldflags=-linkmode=external -mod=readonly -modcacherw" go build -o 'deleteme' ./cmd/slackdump

channels.go

@@ -69,7 +69,6 @@ func (s *Session) getChannels(ctx context.Context, chanTypes []string, cb func(t
 				chans, nextcur, err = s.client.GetConversationsContext(ctx, params)
 			})
 			return err
-
 		}); err != nil {
 			return err
 		}

cmd/slackdump/instruments_test.go

@@ -0,0 +1,17 @@
+package main
+
+import (
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func Test_initTrace(t *testing.T) {
+	t.Run("initialises trace file", func(t *testing.T) {
+		testTraceFile := filepath.Join(t.TempDir(), "trace.out")
+		stop := initTrace(testTraceFile)
+		t.Cleanup(stop)
+		assert.FileExists(t, testTraceFile)
+	})
+}

cmd/slackdump/internal/archive/archive.go

@@ -4,9 +4,14 @@ import (
 	"context"
 	_ "embed"
 	"errors"
+	"io"
 	"log/slog"
+	"os"
+	"path/filepath"
+	"strings"
 	"time"
 
+	"github.com/jmoiron/sqlx"
 	"github.com/rusq/fsadapter"
 
 	"github.com/rusq/slackdump/v3"
@@ -15,6 +20,8 @@ import (
 	"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"
+	"github.com/rusq/slackdump/v3/internal/chunk/dbproc"
+	"github.com/rusq/slackdump/v3/internal/chunk/dbproc/repository"
 	"github.com/rusq/slackdump/v3/internal/chunk/transform/fileproc"
 	"github.com/rusq/slackdump/v3/internal/structures"
 	"github.com/rusq/slackdump/v3/stream"
@@ -40,6 +47,14 @@ func init() {
 var errNoOutput = errors.New("output directory is required")
 
 func RunArchive(ctx context.Context, cmd *base.Command, args []string) error {
+	if cfg.UseChunkFiles {
+		return runChunkArchive(ctx, cmd, args)
+	} else {
+		return runDBArchive(ctx, cmd, args)
+	}
+}
+
+func runChunkArchive(ctx context.Context, _ *base.Command, args []string) error {
 	start := time.Now()
 	list, err := structures.NewEntityList(args)
 	if err != nil {
@@ -67,7 +82,50 @@ func RunArchive(ctx context.Context, cmd *base.Command, args []string) error {
 		base.SetExitStatus(base.SApplicationError)
 		return err
 	}
-	cfg.Log.Info("Recorded workspace data", "filename", cd.Name(), "took", time.Since(start))
+	cfg.Log.Info("Recorded workspace data", "directory", cd.Name(), "took", time.Since(start))
+	return nil
+}
+
+func runDBArchive(ctx context.Context, _ *base.Command, args []string) error {
+	start := time.Now()
+	list, err := structures.NewEntityList(args)
+	if err != nil {
+		base.SetExitStatus(base.SUserError)
+		return err
+	}
+	sess, err := bootstrap.SlackdumpSession(ctx)
+	if err != nil {
+		base.SetExitStatus(base.SInitializationError)
+		return err
+	}
+
+	dirname := cfg.StripZipExt(cfg.Output)
+	if err := os.MkdirAll(dirname, 0o755); err != nil {
+		return err
+	}
+
+	conn, err := sqlx.Open(repository.Driver, filepath.Join(dirname, "slackdump.sqlite"))
+	if err != nil {
+		return err
+	}
+	defer conn.Close()
+
+	ctrl, err := DBController(ctx, conn, sess, dirname)
+	if err != nil {
+		return err
+	}
+
+	defer func() {
+		if err := ctrl.Close(); err != nil {
+			slog.ErrorContext(ctx, "unable to close database controller", "error", err)
+		}
+	}()
+
+	if err := ctrl.Run(ctx, list); err != nil {
+		base.SetExitStatus(base.SApplicationError)
+		return err
+	}
+	cfg.Log.Info("Recorded workspace data", "directory", dirname, "took", time.Since(start))
 
 	return nil
 }
@@ -87,9 +145,63 @@ func NewDirectory(name string) (*chunk.Directory, error) {
 	return cd, nil
 }
 
+func DBController(ctx context.Context, conn *sqlx.DB, sess *slackdump.Session, dirname string, opts ...stream.Option) (RunCloser, error) {
+	lg := cfg.Log
+	dbp, err := dbproc.New(ctx, conn, dbproc.SessionInfo{
+		FromTS:         &time.Time{},
+		ToTS:           &time.Time{},
+		FilesEnabled:   cfg.DownloadFiles,
+		AvatarsEnabled: cfg.DownloadAvatars,
+		Mode:           "archive",
+		Args:           strings.Join(os.Args, "|"),
+	})
+	if err != nil {
+		return nil, err
+	}
+	sopts := []stream.Option{
+		stream.OptLatest(time.Time(cfg.Latest)),
+		stream.OptOldest(time.Time(cfg.Oldest)),
+		stream.OptResultFn(resultLogger(lg)),
+	}
+	sopts = append(sopts, opts...)
+	// start attachment downloader
+	dl := fileproc.NewDownloader(
+		ctx,
+		cfg.DownloadFiles,
+		sess.Client(),
+		fsadapter.NewDirectory(dirname),
+		lg,
+	)
+	// start avatar downloader
+	avdl := fileproc.NewDownloader(
+		ctx,
+		cfg.DownloadAvatars,
+		sess.Client(),
+		fsadapter.NewDirectory(dirname),
+		lg,
+	)
+
+	ctrl, err := control.NewDB(
+		ctx,
+		sess.Stream(sopts...),
+		dbp,
+		control.WithFiler(fileproc.New(dl)),
+		control.WithAvatarProcessor(fileproc.NewAvatarProc(avdl)),
+	)
+	if err != nil {
+		return nil, err
+	}
+	return ctrl, nil
+}
+
+type RunCloser interface {
+	Run(context.Context, *structures.EntityList) error
+	io.Closer
+}
+
 // ArchiveController returns the default archive controller initialised based
 // on global configuration parameters.
-func ArchiveController(ctx context.Context, cd *chunk.Directory, sess *slackdump.Session, opts ...stream.Option) (*control.Controller, error) {
+func ArchiveController(ctx context.Context, cd *chunk.Directory, sess *slackdump.Session, opts ...stream.Option) (*control.DirController, error) {
 	lg := cfg.Log
 
 	sopts := []stream.Option{
@@ -116,7 +228,7 @@ func ArchiveController(ctx context.Context, cd *chunk.Directory, sess *slackdump
 		lg,
 	)
 
-	ctrl := control.New(
+	ctrl := control.NewDir(
 		cd,
 		sess.Stream(sopts...),
 		control.WithLogger(lg),

cmd/slackdump/internal/archive/search.go

@@ -44,7 +44,7 @@ var cmdSearchMessages = &base.Command{
 	Long:        `Searches for messages matching criteria.`,
 	RequireAuth: true,
 	FlagMask:    flagMask | cfg.OmitRecordFilesFlag,
-	Run:         runSearchFn((*control.Controller).SearchMessages),
+	Run:         runSearchFn((*control.DirController).SearchMessages),
 	PrintFlags:  true,
 }
 
@@ -54,7 +54,7 @@ var cmdSearchFiles = &base.Command{
 	Long:        `Searches for messages matching criteria.`,
 	RequireAuth: true,
 	FlagMask:    flagMask,
-	Run:         runSearchFn((*control.Controller).SearchFiles),
+	Run:         runSearchFn((*control.DirController).SearchFiles),
 	PrintFlags:  true,
 }
 
@@ -64,7 +64,7 @@ var cmdSearchAll = &base.Command{
 	Long:        `Records search message and files results matching the given query`,
 	RequireAuth: true,
 	FlagMask:    flagMask,
-	Run:         runSearchFn((*control.Controller).SearchAll),
+	Run:         runSearchFn((*control.DirController).SearchAll),
 	PrintFlags:  true,
 }
 
@@ -78,7 +78,7 @@ func init() {
 
 var ErrNoQuery = errors.New("missing query parameter")
 
-func runSearchFn(fn func(*control.Controller, context.Context, string) error) func(context.Context, *base.Command, []string) error {
+func runSearchFn(fn func(*control.DirController, context.Context, string) error) func(context.Context, *base.Command, []string) error {
 	return func(ctx context.Context, cmd *base.Command, args []string) error {
 		if len(args) == 0 {
 			base.SetExitStatus(base.SInvalidParameters)
@@ -104,7 +104,11 @@ func runSearchFn(fn func(*control.Controller, context.Context, string) error) fu
 		if err != nil {
 			return err
 		}
-		defer ctrl.Close()
+		defer func() {
+			if err := ctrl.Close(); err != nil {
+				cfg.Log.Error("error closing controller", "err", err)
+			}
+		}()
 		defer stop()
 
 		query := strings.Join(args, " ")
@@ -116,7 +120,7 @@ func runSearchFn(fn func(*control.Controller, context.Context, string) error) fu
 	}
 }
 
-func searchController(ctx context.Context, cd *chunk.Directory, sess *slackdump.Session, terms []string) (*control.Controller, func(), error) {
+func searchController(ctx context.Context, cd *chunk.Directory, sess *slackdump.Session, terms []string) (*control.DirController, func(), error) {
 	if len(terms) == 0 {
 		base.SetExitStatus(base.SInvalidParameters)
 		return nil, nil, errors.New("missing query parameter")
@@ -147,14 +151,12 @@ func searchController(ctx context.Context, cd *chunk.Directory, sess *slackdump.
 		sopts = append(sopts, stream.OptFastSearch())
 	}
 
-	var (
-		ctrl = control.New(
-			cd,
-			sess.Stream(sopts...),
-			control.WithLogger(lg),
-			control.WithFiler(fileproc.New(dl)),
-			control.WithFlags(control.Flags{RecordFiles: cfg.RecordFiles}),
-		)
+	ctrl := control.NewDir(
+		cd,
+		sess.Stream(sopts...),
+		control.WithLogger(lg),
+		control.WithFiler(fileproc.New(dl)),
+		control.WithFlags(control.Flags{RecordFiles: cfg.RecordFiles}),
 	)
 	return ctrl, func() { pb.Finish() }, nil
 }

cmd/slackdump/internal/bootstrap/database.go

@@ -0,0 +1,39 @@
+package bootstrap
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/jmoiron/sqlx"
+
+	"github.com/rusq/slackdump/v3/cmd/slackdump/internal/cfg"
+	"github.com/rusq/slackdump/v3/internal/chunk/dbproc"
+	"github.com/rusq/slackdump/v3/internal/chunk/dbproc/repository"
+)
+
+const defFilename = "slackdump.sqlite"
+
+// Database returns the initialised database connection open for writing.
+func Database(dir string, mode string) (*sqlx.DB, dbproc.SessionInfo, error) {
+	dbfile := filepath.Join(dir, defFilename)
+	// wconn is the writer connection
+	wconn, err := sqlx.Open(repository.Driver, dbfile)
+	if err != nil {
+		return nil, dbproc.SessionInfo{}, err
+	}
+	return wconn, sessionInfo(mode), nil
+}
+
+func sessionInfo(mode string) dbproc.SessionInfo {
+	si := dbproc.SessionInfo{
+		FromTS:         (*time.Time)(&cfg.Oldest),
+		ToTS:           (*time.Time)(&cfg.Latest),
+		FilesEnabled:   cfg.DownloadFiles,
+		AvatarsEnabled: cfg.DownloadAvatars,
+		Mode:           mode,
+		Args:           strings.Join(os.Args, "|"),
+	}
+	return si
+}