cosmetic changes

internal/chunk/obfuscate/directory.go

@@ -24,7 +24,7 @@ func DoDir(ctx context.Context, src string, trg string, options ...Option) error
 	for _, optFn := range options {
 		optFn(&opts)
 	}
-	rand.Seed(opts.seed)
+	rand.New(rand.NewSource(opts.seed))
 
 	lg := dlog.FromContext(ctx)
 	files, err := os.ReadDir(src)

internal/chunk/state/state.go

@@ -2,14 +2,17 @@ package state
 
 import (
 	"encoding/json"
+	"errors"
 	"io"
 	"os"
+	"path/filepath"
 	"strconv"
 	"strings"
 	"sync"
 
 	"github.com/rusq/fsadapter"
 	"github.com/rusq/slackdump/v3/internal/fasttime"
+	"github.com/rusq/slackdump/v3/internal/osext"
 )
 
 const Version = 0.1
@@ -74,6 +77,33 @@ func New(filename string) *State {
 	}
 }
 
+// State holds the state of a chunk recording. It contains the filename of the
+// chunk recording file, as well as the path to the downloaded files.
+
+var ErrNoChunkFile = errors.New("no linked chunk file")
+
+// OpenChunks attempts to open the chunk file linked in the State. If the
+// chunk is compressed, it will be decompressed and a temporary file will be
+// created. The temporary file will be removed when the OpenChunks is
+// closed.
+func (st *State) OpenChunks(basePath string) (io.ReadSeekCloser, error) {
+	if st.ChunkFilename == "" {
+		return nil, ErrNoChunkFile
+	}
+	f, err := os.Open(filepath.Join(basePath, st.ChunkFilename))
+	if err != nil {
+		return nil, err
+	}
+	if st.IsCompressed {
+		tf, err := osext.UnGZIP(f)
+		if err != nil {
+			return nil, err
+		}
+		return osext.RemoveOnClose(tf), nil
+	}
+	return f, nil
+}
+
 // AddMessage indexes the message.  It should be called when a message is
 // processed.
 func (s *State) AddMessage(channelID, messageTS string) {

internal/chunk/transform/standard.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"sort"
+	"strings"
 
 	"github.com/rusq/fsadapter"
 	"github.com/rusq/slack"
@@ -151,13 +152,14 @@ func stdConversation(cf *chunk.File, ci *slack.Channel, pipeline pipeline) ([]ty
 	}
 	msgs := make([]types.Message, 0, len(mm))
 	for i := range mm {
-		if mm[i].SubType == structures.SubTypeThreadBroadcast {
+		if strings.EqualFold(mm[i].SubType, structures.SubTypeThreadBroadcast) {
 			// this we don't eat.  Skip thread broadcasts.
 			continue
 		}
 		var sdm types.Message // slackdump message
 		sdm.Message = mm[i]
-		if mm[i].ThreadTimestamp != "" && mm[i].ThreadTimestamp == mm[i].Timestamp && mm[i].LatestReply != structures.LatestReplyNoReplies { // process thread only for parent messages
+		if mm[i].ThreadTimestamp != "" && mm[i].ThreadTimestamp == mm[i].Timestamp && mm[i].LatestReply != structures.LatestReplyNoReplies {
+			// process thread only for parent messages
 			// if there's a thread timestamp, we need to find and add it.
 			thread, err := cf.AllThreadMessages(ci.ID, mm[i].ThreadTimestamp)
 			if err != nil {