Merge pull request #564 from rusq/i560-export-source

Export source fixes

Author: rusq

go.mod

@@ -88,6 +88,7 @@ require (
 	github.com/muesli/cancelreader v0.2.2 // indirect
 	github.com/muesli/termenv v0.16.0 // indirect
 	github.com/ncruces/go-strftime v0.1.9 // indirect
+	github.com/phuslu/lru v1.0.18 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
 	github.com/rivo/uniseg v0.4.7 // indirect

go.sum

@@ -177,6 +177,8 @@ github.com/muesli/termenv v0.16.0 h1:S5AlUN9dENB57rsbnkPyfdGuWIlkmzJjbFf0Tf5FWUc
 github.com/muesli/termenv v0.16.0/go.mod h1:ZRfOIKPFDYQoDFF4Olj7/QJbW60Ol/kL1pU3VfY/Cnk=
 github.com/ncruces/go-strftime v0.1.9 h1:bY0MQC28UADQmHmaF5dgpLmImcShSi2kHU9XLdhx/f4=
 github.com/ncruces/go-strftime v0.1.9/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls=
+github.com/phuslu/lru v1.0.18 h1:ioKRYLym7nv6UmaKHXSR0Z8s2KCEra+mcWcn9zXQnlM=
+github.com/phuslu/lru v1.0.18/go.mod h1:ci5hb8dRIa+2I+KcPl4958OWCg09FxwZCP8InU1L1ME=
 github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c h1:+mdjkGKdHQG3305AYmdv1U2eRNDiU2ErMBj1gwrq8eQ=
 github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c/go.mod h1:7rwL4CYBLnjLxUqIJNnCWiEdr3bn6IUYi15bNlnbCCU=
 github.com/playwright-community/playwright-go v0.5200.0 h1:z/5LGuX2tBrg3ug1HupMXLjIG93f1d2MWdDsNhkMQ9c=

internal/convert/src2enc.go

@@ -16,7 +16,7 @@ import (
 	"github.com/rusq/slackdump/v3/source"
 )
 
-// Source encoder allows to convert any source to a chunked format.
+// SourceEncoder allows to convert any source to a chunked format.
 type SourceEncoder struct {
 	src  source.Sourcer
 	enc  chunk.Encoder

internal/structures/conversation.go

@@ -2,9 +2,10 @@ package structures
 
 import "github.com/rusq/slack"
 
-// IsThreadStart check if the message is a lead message of a thread.
+// IsThreadStart check if the message is a lead message of a thread and the
+// thread is a non-empty thread.
 func IsThreadStart(m *slack.Message) bool {
-	return m.ThreadTimestamp != "" && m.Timestamp == m.ThreadTimestamp
+	return m.ThreadTimestamp != "" && m.Timestamp == m.ThreadTimestamp && !IsEmptyThread(m)
 }
 
 // IsEmptyThread checks if the message is a thread with no replies.

notes.md

@@ -0,0 +1,29 @@
+Issue # 560
+-----------
+
+Convert from database to export.
+Convert from export to database.
+Run comparison.
+
+Run convert of the tmp/reduced to database.
+```
+$ sd convert -f database tmp/reduced
+```
+
+in tmp/compare, start sqlite
+run
+```sql
+	attach database 'conv.sqlite' as conv;
+	attach database 'orig.sqlite' as orig;
+```
+
+Output of the missing data:
+```sql
+	with missing as (select id from orig.message where id not in (select id from conv.message))
+	select id,chunk_id,channel_id,ts,parent_id,thread_ts,is_parent from orig.message where id in (select * from missing);
+```
+
+comparison of IDs:
+```sql
+select o.id,o.cnt,c.id,c.cnt from (select id,count(1) cnt from orig.message group by id) o left join (select id,count(1) cnt from conv.message group by id) c on o.id=c.id where o.cnt <> coalesce(c.cnt,-1);
+```

source/export.go

@@ -9,13 +9,13 @@ import (
 	"iter"
 	"log/slog"
 	"path"
+	"runtime/trace"
 	"time"
 
-	"github.com/rusq/slackdump/v3/internal/chunk"
-
 	"github.com/rusq/slack"
 
 	"github.com/rusq/slackdump/v3/export"
+	"github.com/rusq/slackdump/v3/internal/chunk"
 	"github.com/rusq/slackdump/v3/internal/structures"
 )
 
@@ -28,8 +28,13 @@ type Export struct {
 	idx       structures.ExportIndex
 	files     Storage
 	avatars   Storage
+	cache     *threadCache
 }
 
+const cacheSz = 1 << 20
+
+// OpenExport opens a Slack export with the given name from the filesystem
+// fsys.
 func OpenExport(fsys fs.FS, name string) (*Export, error) {
 	var idx structures.ExportIndex
 	if err := idx.Unmarshal(fsys); err != nil {
@@ -44,6 +49,7 @@ func OpenExport(fsys fs.FS, name string) (*Export, error) {
 		chanNames: make(map[string]string, len(chans)),
 		files:     NoStorage{},
 		avatars:   NoStorage{},
+		cache:     newThreadCache(cacheSz),
 	}
 	// initialise channels for quick lookup
 	for _, ch := range z.channels {
@@ -95,15 +101,46 @@ func (e *Export) Type() Flags {
 	return FExport
 }
 
+// buildThreadCache walks all messages in the channel with the given name and
+// indexes all threads for faster lookup.
+func (e *Export) buildThreadCache(ctx context.Context, name string) error {
+	lg := slog.With("channel_name", name)
+	lg.Debug("building thread cache")
+	var n int
+	if err := walkDir(e.fs, name, func(file string) error {
+		if err := yieldFileContents(ctx, e.fs, file, func(m slack.Message, err error) bool {
+			if err != nil {
+				return false
+			}
+			if (structures.IsThreadStart(&m) && !structures.IsEmptyThread(&m)) || structures.IsThreadMessage(&m.Msg) {
+				if err := e.cache.Update(ctx, name, m.ThreadTimestamp, file); err != nil {
+					slog.ErrorContext(ctx, "error updating cache", "error", err)
+				}
+				n++
+			}
+			return true
+		}); err != nil {
+			return err
+		}
+		return nil
+	}); err != nil {
+		return err
+	}
+	slog.DebugContext(ctx, "caching completed", "thread_count", n)
+	return nil
+}
+
 // AllMessages returns all channel messages without thread messages.
-func (e *Export) AllMessages(_ context.Context, channelID string) (iter.Seq2[slack.Message, error], error) {
-	it, err := e.walkChannelMessages(channelID)
+func (e *Export) AllMessages(ctx context.Context, channelID string) (iter.Seq2[slack.Message, error], error) {
+	name, err := e.nameByID(channelID)
 	if err != nil {
-		if errors.Is(err, fs.ErrNotExist) {
-			return nil, ErrNotFound
-		}
 		return nil, err
 	}
+
+	if err := e.buildThreadCache(ctx, name); err != nil {
+		return nil, err
+	}
+	it := e.walkChannelMessages(ctx, name)
 	return func(yield func(slack.Message, error) bool) {
 		for m, err := range it {
 			if err != nil {
@@ -122,63 +159,164 @@ func (e *Export) AllMessages(_ context.Context, channelID string) (iter.Seq2[sla
 	}, nil
 }
 
-func (e *Export) walkChannelMessages(channelID string) (iter.Seq2[slack.Message, error], error) {
-	name, ok := e.chanNames[channelID]
-	if !ok {
-		return nil, fmt.Errorf("%w: %s", fs.ErrNotExist, channelID)
+// yieldFileContents is meant to work with export json files and will call
+// yield function for every message in the file. It expects to be called by
+// fs.WalkDir function, therefore when the yield function returns false (stop
+// iteration), it returns `fs.SkipAll` error.  If calling this function not
+// from the Walk function, this error indicates that file iteration should
+// stop.
+func yieldFileContents(ctx context.Context, fsys fs.FS, file string, yield func(slack.Message, error) bool) error {
+	em, err := unmarshal[[]export.ExportMessage](fsys, file)
+	if err != nil {
+		var jsonErr *json.SyntaxError
+		if errors.As(err, &jsonErr) {
+			slog.WarnContext(ctx, "skipping broken file", "pth", file, "err", err)
+			return nil
+		}
+		return err
+	}
+	for i, m := range em {
+		if m.Msg == nil {
+			slog.DebugContext(ctx, "skipping an empty message", "pth", file, "index", i)
+			continue
+		}
+		sm := slack.Message{Msg: *m.Msg}
+		if !yield(sm, nil) {
+			return fs.SkipAll
+		}
 	}
-	_, err := fs.Stat(e.fs, name)
+	return nil
+}
+
+// fullScanIter is the message iterator that always scans all messages and
+// populates the cache with discovered threads.
+type fullScanIter struct {
+	ctx  context.Context
+	name string
+	fs   fs.FS
+}
+
+func newFullScanIter(ctx context.Context, fs fs.FS, chanName string) *fullScanIter {
+	return &fullScanIter{
+		ctx:  ctx,
+		name: chanName,
+		fs:   fs,
+	}
+}
+
+// Iter iterates through all messages for the given channel name. It
+// updates the cache with discovered threads.
+func (w *fullScanIter) Iter(yield func(slack.Message, error) bool) {
+	ctx, task := trace.NewTask(w.ctx, "full_scan_iter")
+	defer task.End()
+	err := walkDir(w.fs, w.name, func(file string) error {
+		if err := yieldFileContents(ctx, w.fs, file, yield); err != nil {
+			return err
+		}
+		return nil
+	})
 	if err != nil {
-		return nil, fmt.Errorf("%w: %s", fs.ErrNotExist, name)
+		yield(slack.Message{}, err)
+		return
 	}
-	iterFn := func(yield func(slack.Message, error) bool) {
-		err := fs.WalkDir(e.fs, name, func(pth string, d fs.DirEntry, err error) error {
-			if err != nil {
-				return err
-			}
-			if d.IsDir() && pth != name {
-				return fs.SkipDir
-			}
-			if path.Ext(pth) != ".json" {
-				return nil
-			}
-			// read the file
-			em, err := unmarshal[[]export.ExportMessage](e.fs, pth)
-			if err != nil {
-				var jsonErr *json.SyntaxError
-				if errors.As(err, &jsonErr) {
-					slog.Default().Debug("skipping a broken file", "pth", pth, "err", err)
-					return nil
-				}
-				return err
-			}
-			for i, m := range em {
-				if m.Msg == nil {
-					slog.Default().Debug("skipping an empty message", "pth", pth, "index", i)
-					continue
-				}
-				sm := slack.Message{Msg: *m.Msg}
-				if !yield(sm, nil) {
-					return fs.SkipAll
-				}
-			}
-			return nil
-		})
+}
+
+// walkDir walks through the directory with given name on the filesystem fsys,
+// calling the callback function cb for every JSON file it encounters.
+func walkDir(fsys fs.FS, dirName string, cb func(file string) error) error {
+	err := fs.WalkDir(fsys, dirName, func(file string, d fs.DirEntry, err error) error {
 		if err != nil {
+			return err
+		}
+		if d.IsDir() && file != dirName {
+			return fs.SkipDir
+		}
+		if path.Ext(file) != ".json" {
+			return nil
+		}
+		return cb(file)
+	})
+	return err
+}
+
+// fileListIter is meant to reduce the scope of iteration to the given file
+// list.
+type fileListIter struct {
+	ctx   context.Context
+	fs    fs.FS
+	files []string
+}
+
+func (w *fileListIter) Iter(yield func(slack.Message, error) bool) {
+	ctx, task := trace.NewTask(w.ctx, "file_list_iter")
+	defer task.End()
+	for _, file := range w.files {
+		if err := yieldFileContents(ctx, w.fs, file, yield); err != nil {
+			if errors.Is(err, fs.SkipAll) {
+				// bail out if instructed
+				return
+			}
 			yield(slack.Message{}, err)
+			return
 		}
 	}
-	return iterFn, nil
 }
 
-func (e *Export) AllThreadMessages(_ context.Context, channelID, threadID string) (iter.Seq2[slack.Message, error], error) {
-	it, err := e.walkChannelMessages(channelID)
+// nameByID returns a channel name (directory name) by the channelID.
+// It ensures that the directory exists. It will return ErrNotFound
+// if it doesn't find the channel or it's directory.
+func (e *Export) nameByID(channelID string) (string, error) {
+	name, ok := e.chanNames[channelID]
+	if !ok {
+		return "", fmt.Errorf("%w: %s", ErrNotFound, channelID)
+	}
+	if fi, err := fs.Stat(e.fs, name); err != nil {
+		return "", fmt.Errorf("%w: %s", ErrNotFound, name)
+	} else if !fi.IsDir() {
+		return "", fmt.Errorf("%s is not a directory", name)
+	}
+	return name, nil
+}
+
+func (e *Export) walkChannelMessages(ctx context.Context, name string) iter.Seq2[slack.Message, error] {
+	return newFullScanIter(ctx, e.fs, name).Iter
+}
+
+var errNotInCache = errors.New("channel not in cache")
+
+func (e *Export) walkCachedThreads(ctx context.Context, channelName, threadID string) (iter.Seq2[slack.Message, error], error) {
+	if !e.cache.Exists(channelName) {
+		return nil, fmt.Errorf("channel: %w", errNotInCache)
+	}
+	// get all files for the thread.
+	files, ok := e.cache.Get(channelName, threadID)
+	if !ok {
+		return nil, fmt.Errorf("thread: %w", errNotInCache)
+	}
+	fli := fileListIter{ctx, e.fs, files}
+	return fli.Iter, nil
+}
+
+// AllThreadMessages returns all thread messages for the channelID:threadID. If the thread
+// is contained in the cache, it will iterate only through the files that contain the thread
+// messages, otherwise it will iterate through all messages in the channel and extract the thread
+// messages. Call [buildThreadCache] for the channelID, before calling this
+// method to speed up search.
+func (e *Export) AllThreadMessages(ctx context.Context, channelID, threadID string) (iter.Seq2[slack.Message, error], error) {
+	// try cached first
+	name, err := e.nameByID(channelID)
 	if err != nil {
-		if errors.Is(err, fs.ErrNotExist) {
-			return nil, ErrNotFound
-		}
 		return nil, err
 	}
+	lg := slog.With("channel_name", name, "channel_id", channelID, "thread_ts", threadID)
+	it, err := e.walkCachedThreads(ctx, name, threadID)
+	if err != nil {
+		if !errors.Is(err, errNotInCache) {
+			return nil, err
+		}
+		lg.WarnContext(ctx, "cache not available, initiating full scan", "err", err)
+		it = e.walkChannelMessages(ctx, name)
+	}
 	iterFn := func(yield func(slack.Message, error) bool) {
 		for m, err := range it {
 			if err != nil {