From f3cc5df3a4b86936e03a75feada712c7e3f10e88 Mon Sep 17 00:00:00 2001
From: Rustam Gilyazov <16064414+rusq@users.noreply.github.com>
Date: Sat, 22 Feb 2025 08:10:48 +1000
Subject: [PATCH] generalising
---
cmd/slackdump/internal/convertcmd/convert.go | 4 +-
cmd/slackdump/internal/diag/hydrate.go | 17 ++++--
.../internal/diag/hydrate_mock_test.go | 9 +--
cmd/slackdump/internal/export/v3.go | 4 +-
internal/chunk/db_compat.go | 18 ++++++
internal/chunk/dbproc/source.go | 24 ++++++--
internal/chunk/dbproc/source_test.go | 1 +
internal/chunk/transform/export.go | 55 +++++++++--------
internal/chunk/transform/export_test.go | 6 +-
internal/convert/chunkexp.go | 16 +++--
internal/convert/chunkexp_test.go | 16 ++---
internal/convert/dbexp.go | 1 -
internal/source/dump_test.go | 59 +++++++++++++++++++
internal/testutil/iter.go | 15 +++++
internal/viewer/handlers.go | 33 ++++-------
internal/viewer/handlers_test.go | 1 +
internal/viewer/template.go | 10 ++--
internal/viewer/templates/index.html | 24 ++++----
types/message.go | 9 +--
19 files changed, 224 insertions(+), 98 deletions(-)
create mode 100644 internal/chunk/db_compat.go
create mode 100644 internal/chunk/dbproc/source_test.go
delete mode 100644 internal/convert/dbexp.go
create mode 100644 internal/source/dump_test.go
create mode 100644 internal/testutil/iter.go
create mode 100644 internal/viewer/handlers_test.go
diff --git a/cmd/slackdump/internal/convertcmd/convert.go b/cmd/slackdump/internal/convertcmd/convert.go
index ad7f279c..77596d3e 100644
--- a/cmd/slackdump/internal/convertcmd/convert.go
+++ b/cmd/slackdump/internal/convertcmd/convert.go
@@ -115,6 +115,7 @@ func chunk2export(ctx context.Context, src, trg string, cflg convertflags) error
return err
}
defer cd.Close()
+
fsa, err := fsadapter.New(trg)
if err != nil {
return err
@@ -131,8 +132,9 @@ func chunk2export(ctx context.Context, src, trg string, cflg convertflags) error
includeAvatars = cflg.withAvatars && (st&source.FAvatars != 0)
)
+ s := source.NewChunkDir(cd, true)
cvt := convert.NewChunkToExport(
- cd,
+ s,
fsa,
convert.WithIncludeFiles(includeFiles),
convert.WithIncludeAvatars(includeAvatars),
diff --git a/cmd/slackdump/internal/diag/hydrate.go b/cmd/slackdump/internal/diag/hydrate.go
index 2e3c4386..106aad5c 100644
--- a/cmd/slackdump/internal/diag/hydrate.go
+++ b/cmd/slackdump/internal/diag/hydrate.go
@@ -6,6 +6,7 @@ import (
"errors"
"fmt"
"io"
+ "iter"
"log/slog"
"net/http"
"net/url"
@@ -179,8 +180,8 @@ func download(ctx context.Context, archive, target string, dry bool) error {
//go:generate mockgen -destination=hydrate_mock_test.go -package=diag -source hydrate.go sourcer
type sourcer interface {
Channels(ctx context.Context) ([]slack.Channel, error)
- AllMessages(ctx context.Context, channelID string) ([]slack.Message, error)
- AllThreadMessages(ctx context.Context, channelID, threadTimestamp string) ([]slack.Message, error)
+ AllMessages(ctx context.Context, channelID string) (iter.Seq2[slack.Message, error], error)
+ AllThreadMessages(ctx context.Context, channelID, threadTimestamp string) (iter.Seq2[slack.Message, error], error)
}
func downloadFiles(ctx context.Context, d downloader.GetFiler, trg fsadapter.FS, src sourcer) error {
@@ -202,18 +203,24 @@ func downloadFiles(ctx context.Context, d downloader.GetFiler, trg fsadapter.FS,
if err != nil {
return fmt.Errorf("error reading messages in channel %s: %w", ch.ID, err)
}
- for _, m := range msgs {
+ for m, err := range msgs {
+ if err != nil {
+ return fmt.Errorf("error reading message in channel %s: %w", ch.ID, err)
+ }
if len(m.Files) > 0 {
if err := proc.Files(ctx, &ch, m, m.Files); err != nil {
return fmt.Errorf("error processing files in message %s: %w", m.Timestamp, err)
}
}
if structures.IsThreadStart(&m) {
- tm, err := src.AllThreadMessages(ctx, ch.ID, m.ThreadTimestamp)
+ itTm, err := src.AllThreadMessages(ctx, ch.ID, m.ThreadTimestamp)
if err != nil {
return fmt.Errorf("error reading thread messages for message %s in channel %s: %w", m.Timestamp, ch.ID, err)
}
- for _, tm := range tm {
+ for tm, err := range itTm {
+ if err != nil {
+ return fmt.Errorf("error reading thread message %s in channel %s: %w", tm.Timestamp, ch.ID, err)
+ }
if len(tm.Files) > 0 {
if err := proc.Files(ctx, &ch, tm, tm.Files); err != nil {
return fmt.Errorf("error processing files in thread message %s: %w", tm.Timestamp, err)
diff --git a/cmd/slackdump/internal/diag/hydrate_mock_test.go b/cmd/slackdump/internal/diag/hydrate_mock_test.go
index 58c8aa07..03d42b59 100644
--- a/cmd/slackdump/internal/diag/hydrate_mock_test.go
+++ b/cmd/slackdump/internal/diag/hydrate_mock_test.go
@@ -11,6 +11,7 @@ package diag
import (
context "context"
+ iter "iter"
reflect "reflect"
slack "github.com/rusq/slack"
@@ -42,10 +43,10 @@ func (m *Mocksourcer) EXPECT() *MocksourcerMockRecorder {
}
// AllMessages mocks base method.
-func (m *Mocksourcer) AllMessages(ctx context.Context, channelID string) ([]slack.Message, error) {
+func (m *Mocksourcer) AllMessages(ctx context.Context, channelID string) (iter.Seq2[slack.Message, error], error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "AllMessages", ctx, channelID)
- ret0, _ := ret[0].([]slack.Message)
+ ret0, _ := ret[0].(iter.Seq2[slack.Message, error])
ret1, _ := ret[1].(error)
return ret0, ret1
}
@@ -57,10 +58,10 @@ func (mr *MocksourcerMockRecorder) AllMessages(ctx, channelID any) *gomock.Call
}
// AllThreadMessages mocks base method.
-func (m *Mocksourcer) AllThreadMessages(ctx context.Context, channelID, threadTimestamp string) ([]slack.Message, error) {
+func (m *Mocksourcer) AllThreadMessages(ctx context.Context, channelID, threadTimestamp string) (iter.Seq2[slack.Message, error], error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "AllThreadMessages", ctx, channelID, threadTimestamp)
- ret0, _ := ret[0].([]slack.Message)
+ ret0, _ := ret[0].(iter.Seq2[slack.Message, error])
ret1, _ := ret[1].(error)
return ret0, ret1
}
diff --git a/cmd/slackdump/internal/export/v3.go b/cmd/slackdump/internal/export/v3.go
index 062bb517..dec335f6 100644
--- a/cmd/slackdump/internal/export/v3.go
+++ b/cmd/slackdump/internal/export/v3.go
@@ -18,6 +18,7 @@ import (
"github.com/rusq/slackdump/v3/internal/chunk/control"
"github.com/rusq/slackdump/v3/internal/chunk/transform"
"github.com/rusq/slackdump/v3/internal/chunk/transform/fileproc"
+ "github.com/rusq/slackdump/v3/internal/source"
"github.com/rusq/slackdump/v3/internal/structures"
"github.com/rusq/slackdump/v3/stream"
)
@@ -48,7 +49,8 @@ func export(ctx context.Context, sess *slackdump.Session, fsa fsadapter.FS, list
return fn(m)
}
}
- conv := transform.NewExpConverter(chunkdir, fsa, transform.ExpWithMsgUpdateFunc(updFn()))
+ src := source.NewChunkDir(chunkdir, false)
+ conv := transform.NewExpConverter(src, fsa, transform.ExpWithMsgUpdateFunc(updFn()))
tf := transform.NewExportCoordinator(ctx, conv, transform.WithBufferSize(1000))
defer tf.Close()
diff --git a/internal/chunk/db_compat.go b/internal/chunk/db_compat.go
new file mode 100644
index 00000000..7bc9206a
--- /dev/null
+++ b/internal/chunk/db_compat.go
@@ -0,0 +1,18 @@
+package chunk
+
+import (
+ "context"
+
+ "github.com/rusq/slack"
+)
+
+// db source compatibility layer
+
+func (d *Directory) ChannelInfo(ctx context.Context, id string) (*slack.Channel, error) {
+ f, err := d.Open(ToFileID(id, "", false))
+ if err != nil {
+ return nil, err
+ }
+ defer f.Close()
+ return f.ChannelInfo(id)
+}
diff --git a/internal/chunk/dbproc/source.go b/internal/chunk/dbproc/source.go
index 9c557b9e..9cf7c28b 100644
--- a/internal/chunk/dbproc/source.go
+++ b/internal/chunk/dbproc/source.go
@@ -74,6 +74,22 @@ type valuer[T any] interface {
Val() (T, error)
}
+func valueIter[T any, D valuer[T]](it iter.Seq2[D, error]) iter.Seq2[T, error] {
+ iterFn := func(yield func(T, error) bool) {
+ for c, err := range it {
+ if err != nil {
+ var t T
+ yield(t, err)
+ return
+ }
+ if !yield(c.Val()) {
+ return
+ }
+ }
+ }
+ return iterFn
+}
+
func collect[T any, D valuer[T]](it iter.Seq2[D, error], sz int) ([]T, error) {
vs := make([]T, 0, sz)
for c, err := range it {
@@ -89,7 +105,7 @@ func collect[T any, D valuer[T]](it iter.Seq2[D, error], sz int) ([]T, error) {
return vs, nil
}
-func (s *Source) AllMessages(ctx context.Context, channelID string) ([]slack.Message, error) {
+func (s *Source) AllMessages(ctx context.Context, channelID string) (iter.Seq2[slack.Message, error], error) {
tx, err := s.conn.BeginTxx(ctx, &sql.TxOptions{ReadOnly: true})
if err != nil {
return nil, err
@@ -101,10 +117,10 @@ func (s *Source) AllMessages(ctx context.Context, channelID string) ([]slack.Mes
if err != nil {
return nil, err
}
- return collect(it, preallocSz)
+ return valueIter(it), nil
}
-func (s *Source) AllThreadMessages(ctx context.Context, channelID, threadID string) ([]slack.Message, error) {
+func (s *Source) AllThreadMessages(ctx context.Context, channelID, threadID string) (iter.Seq2[slack.Message, error], error) {
tx, err := s.conn.BeginTxx(ctx, &sql.TxOptions{ReadOnly: true})
if err != nil {
return nil, err
@@ -116,7 +132,7 @@ func (s *Source) AllThreadMessages(ctx context.Context, channelID, threadID stri
if err != nil {
return nil, err
}
- return collect(it, preallocSz)
+ return valueIter(it), nil
}
func (s *Source) ChannelInfo(ctx context.Context, channelID string) (*slack.Channel, error) {
diff --git a/internal/chunk/dbproc/source_test.go b/internal/chunk/dbproc/source_test.go
new file mode 100644
index 00000000..6605a4fa
--- /dev/null
+++ b/internal/chunk/dbproc/source_test.go
@@ -0,0 +1 @@
+package dbproc
diff --git a/internal/chunk/transform/export.go b/internal/chunk/transform/export.go
index aeca1158..55fcedac 100644
--- a/internal/chunk/transform/export.go
+++ b/internal/chunk/transform/export.go
@@ -10,7 +10,6 @@ import (
"runtime/trace"
"sort"
"sync/atomic"
- "time"
"github.com/rusq/fsadapter"
"github.com/rusq/slack"
@@ -18,6 +17,7 @@ import (
"github.com/rusq/slackdump/v3/export"
"github.com/rusq/slackdump/v3/internal/chunk"
"github.com/rusq/slackdump/v3/internal/fasttime"
+ "github.com/rusq/slackdump/v3/internal/source"
"github.com/rusq/slackdump/v3/internal/structures"
"github.com/rusq/slackdump/v3/types"
)
@@ -39,15 +39,15 @@ func ExpWithUsers(users []slack.User) ExpCvtOption {
}
type ExpConverter struct {
- cd *chunk.Directory
+ src source.Sourcer
fsa fsadapter.FS
users atomic.Value
msgFunc []msgUpdFunc
}
-func NewExpConverter(cd *chunk.Directory, fsa fsadapter.FS, opt ...ExpCvtOption) *ExpConverter {
+func NewExpConverter(src source.Sourcer, fsa fsadapter.FS, opt ...ExpCvtOption) *ExpConverter {
e := &ExpConverter{
- cd: cd,
+ src: src,
fsa: fsa,
}
for _, o := range opt {
@@ -83,27 +83,20 @@ func (e *ExpConverter) Convert(ctx context.Context, id chunk.FileID) error {
lg.DebugContext(ctx, "transforming channel", "id", id, "user_count", userCnt)
}
- // load the chunk file
- cf, err := e.cd.Open(id)
- if err != nil {
- return fmt.Errorf("error opening chunk file %q: %w", id, err)
- }
- defer cf.Close()
-
channelID, _ := id.Split()
- ci, err := cf.ChannelInfo(channelID)
+ ci, err := e.src.ChannelInfo(ctx, channelID)
if err != nil {
return fmt.Errorf("error reading channel info for %q: %w", id, err)
}
- if err := e.writeMessages(ctx, cf, ci); err != nil {
+ if err := e.writeMessages(ctx, ci); err != nil {
return err
}
return nil
}
-func (e *ExpConverter) writeMessages(ctx context.Context, pl *chunk.File, ci *slack.Channel) error {
+func (e *ExpConverter) writeMessages(ctx context.Context, ci *slack.Channel) error {
lg := slog.With("in", "writeMessages", "channel", ci.ID)
uidx := types.Users(e.getUsers()).IndexByID()
trgdir := ExportChanName(ci)
@@ -111,7 +104,18 @@ func (e *ExpConverter) writeMessages(ctx context.Context, pl *chunk.File, ci *sl
mm := make([]export.ExportMessage, 0, 100)
var prevDt string
var currDt string
- if err := pl.Sorted(ctx, false, func(ts time.Time, m *slack.Message) error {
+ it, err := e.src.AllMessages(ctx, ci.ID)
+ if err != nil {
+ return fmt.Errorf("error getting messages for %q: %w", ci.ID, err)
+ }
+ for m, err := range it {
+ if err != nil {
+ return fmt.Errorf("error reading message: %w", err)
+ }
+ ts, err := structures.ParseSlackTS(m.Timestamp)
+ if err != nil {
+ return fmt.Errorf("error parsing timestamp: %w", err)
+ }
currDt = ts.Format("2006-01-02")
if currDt != prevDt || prevDt == "" {
if prevDt != "" {
@@ -126,10 +130,10 @@ func (e *ExpConverter) writeMessages(ctx context.Context, pl *chunk.File, ci *sl
// the "thread" is only used to collect statistics. Thread messages
// are passed by Sorted and written as a normal course of action.
var thread []slack.Message
- if structures.IsThreadStart(m) && m.LatestReply != structures.LatestReplyNoReplies {
+ if structures.IsThreadStart(&m) && m.LatestReply != structures.LatestReplyNoReplies {
// get the thread for the initial thread message only.
var err error
- thread, err = pl.AllThreadMessages(ci.ID, m.ThreadTimestamp)
+ itTm, err := e.src.AllThreadMessages(ctx, ci.ID, m.ThreadTimestamp)
if err != nil {
if !errors.Is(err, chunk.ErrNotFound) {
return fmt.Errorf("error getting thread messages for %q: %w", ci.ID+":"+m.ThreadTimestamp, err)
@@ -139,19 +143,24 @@ func (e *ExpConverter) writeMessages(ctx context.Context, pl *chunk.File, ci *sl
lg.Warn("not an error, possibly deleted thread not found in chunk file", "slack_link", ci.ID+":"+m.ThreadTimestamp)
}
}
+ thread = make([]slack.Message, 0, 10)
+ for tm, err := range itTm {
+ if err != nil {
+ return fmt.Errorf("error reading thread message: %w", err)
+ }
+ thread = append(thread, tm)
+ }
}
// apply all message functions.
for _, fn := range e.msgFunc {
- if err := fn(ci, m); err != nil {
+ if err := fn(ci, &m); err != nil {
return fmt.Errorf("error updating message: %w", err)
}
}
- mm = append(mm, *toExportMessage(m, thread, uidx[m.User]))
+ mm = append(mm, *toExportMessage(&m, thread, uidx[m.User]))
return nil
- }); err != nil {
- return fmt.Errorf("sorted callback error: %w", err)
}
// flush the last day.
@@ -267,11 +276,11 @@ func ExportChanName(ch *slack.Channel) string {
// once all transformations are done, because it might require to read channel
// files.
func (t *ExpConverter) WriteIndex(ctx context.Context) error {
- wsp, err := t.cd.WorkspaceInfo()
+ wsp, err := t.src.WorkspaceInfo(ctx)
if err != nil {
return fmt.Errorf("failed to get the workspace info: %w", err)
}
- chans, err := t.cd.Channels(ctx)
+ chans, err := t.src.Channels(ctx)
if err != nil {
return fmt.Errorf("error indexing channels: %w", err)
}
diff --git a/internal/chunk/transform/export_test.go b/internal/chunk/transform/export_test.go
index cc82c509..486f2340 100644
--- a/internal/chunk/transform/export_test.go
+++ b/internal/chunk/transform/export_test.go
@@ -12,6 +12,7 @@ import (
"github.com/rusq/slackdump/v3/internal/chunk"
"github.com/rusq/slackdump/v3/internal/fixtures"
+ "github.com/rusq/slackdump/v3/internal/source"
)
func Test_transform(t *testing.T) {
@@ -53,8 +54,9 @@ func Test_transform(t *testing.T) {
t.Fatal(err)
}
defer cd.Close()
+ src := source.NewChunkDir(cd, true)
cvt := ExpConverter{
- cd: cd,
+ src: src,
fsa: tt.args.fsa,
}
if err := cvt.Convert(tt.args.ctx, chunk.FileID(tt.args.id)); (err != nil) != tt.wantErr {
@@ -71,7 +73,6 @@ func TestExpConverter_getUsers(t *testing.T) {
return &v
}
type fields struct {
- cd *chunk.Directory
fsa fsadapter.FS
users atomic.Value
msgFunc []msgUpdFunc
@@ -99,7 +100,6 @@ func TestExpConverter_getUsers(t *testing.T) {
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
e := &ExpConverter{
- cd: tt.fields.cd,
fsa: tt.fields.fsa,
users: tt.fields.users,
msgFunc: tt.fields.msgFunc,
diff --git a/internal/convert/chunkexp.go b/internal/convert/chunkexp.go
index 898bf4df..dadbe5e1 100644
--- a/internal/convert/chunkexp.go
+++ b/internal/convert/chunkexp.go
@@ -17,6 +17,7 @@ import (
"github.com/rusq/slackdump/v3/internal/chunk"
"github.com/rusq/slackdump/v3/internal/chunk/transform"
"github.com/rusq/slackdump/v3/internal/chunk/transform/fileproc"
+ "github.com/rusq/slackdump/v3/internal/source"
)
const (
@@ -32,7 +33,8 @@ var (
// is not usable.
type ChunkToExport struct {
// src is the source directory with chunks
- src *chunk.Directory
+ // src *chunk.Directory
+ src source.Sourcer
// trg is the target FS for the export
trg fsadapter.FS
opts options
@@ -45,7 +47,7 @@ type ChunkToExport struct {
avtrresult chan copyresult
}
-func NewChunkToExport(src *chunk.Directory, trg fsadapter.FS, opt ...C2EOption) *ChunkToExport {
+func NewChunkToExport(src source.Sourcer, trg fsadapter.FS, opt ...C2EOption) *ChunkToExport {
c := &ChunkToExport{
src: src,
trg: trg,
@@ -77,13 +79,9 @@ func (c *ChunkToExport) Validate() error {
return fmt.Errorf("convert: %w", err)
}
// users chunk is required
- if fi, err := c.src.Stat(chunk.FUsers); err != nil {
- return fmt.Errorf("users chunk: %w", err)
- } else if fi.Size() == 0 {
- return fmt.Errorf("users chunk: %w", ErrEmptyChunk)
+ if _, err := c.src.Users(context.Background()); err != nil {
+ return fmt.Errorf("convert: error getting users: %w", err)
}
- // we are not checking for channels chunk because channel information will
- // be stored in each of the chunk files, and we can collect it from there.
return nil
}
@@ -118,7 +116,7 @@ func (c *ChunkToExport) Convert(ctx context.Context) error {
if err != nil {
return err
}
- users, err := c.src.Users()
+ users, err := c.src.Users(ctx)
if err != nil {
return err
}
diff --git a/internal/convert/chunkexp_test.go b/internal/convert/chunkexp_test.go
index 69f73a45..a0bb34ae 100644
--- a/internal/convert/chunkexp_test.go
+++ b/internal/convert/chunkexp_test.go
@@ -12,6 +12,7 @@ import (
"github.com/rusq/slackdump/v3/internal/chunk"
"github.com/rusq/slackdump/v3/internal/fixtures"
+ "github.com/rusq/slackdump/v3/internal/source"
)
const (
@@ -28,10 +29,11 @@ func TestChunkToExport_Validate(t *testing.T) {
t.Fatal(err)
}
defer srcDir.Close()
+ src := source.NewChunkDir(srcDir, true)
testTrgDir := t.TempDir()
type fields struct {
- Src *chunk.Directory
+ Src source.Sourcer
Trg fsadapter.FS
opts options
UploadDir string
@@ -43,11 +45,11 @@ func TestChunkToExport_Validate(t *testing.T) {
}{
{"empty", fields{}, true},
{"no source", fields{Trg: fsadapter.NewDirectory(testTrgDir)}, true},
- {"no target", fields{Src: srcDir}, true},
+ {"no target", fields{Src: src}, true},
{
"valid, no files",
fields{
- Src: srcDir,
+ Src: src,
Trg: fsadapter.NewDirectory(testTrgDir),
opts: options{
includeFiles: false,
@@ -58,7 +60,7 @@ func TestChunkToExport_Validate(t *testing.T) {
{
"valid, include files, but no location functions",
fields{
- Src: srcDir,
+ Src: src,
Trg: fsadapter.NewDirectory(testTrgDir),
opts: options{
includeFiles: true,
@@ -69,7 +71,7 @@ func TestChunkToExport_Validate(t *testing.T) {
{
"valid, include files, with location functions",
fields{
- Src: srcDir,
+ Src: src,
Trg: fsadapter.NewDirectory(testTrgDir),
opts: options{
includeFiles: true,
@@ -117,8 +119,8 @@ func TestChunkToExport_Convert(t *testing.T) {
t.Fatal(err)
}
defer fsa.Close()
-
- c := NewChunkToExport(cd, fsa, WithIncludeFiles(true))
+ src := source.NewChunkDir(cd, true)
+ c := NewChunkToExport(src, fsa, WithIncludeFiles(true))
ctx := context.Background()
c.lg = testLogger
diff --git a/internal/convert/dbexp.go b/internal/convert/dbexp.go
deleted file mode 100644
index 233bcded..00000000
--- a/internal/convert/dbexp.go
+++ /dev/null
@@ -1 +0,0 @@
-package convert
diff --git a/internal/source/dump_test.go b/internal/source/dump_test.go
new file mode 100644
index 00000000..6c96c522
--- /dev/null
+++ b/internal/source/dump_test.go
@@ -0,0 +1,59 @@
+package source
+
+import (
+ "testing"
+
+ "github.com/rusq/slack"
+ "github.com/stretchr/testify/assert"
+
+ "github.com/rusq/slackdump/v3/internal/testutil"
+ "github.com/rusq/slackdump/v3/types"
+)
+
+func Test_convertMessages(t *testing.T) {
+ type args struct {
+ cm []types.Message
+ }
+ tests := []struct {
+ name string
+ args args
+ want []testutil.IterVal[slack.Message, error]
+ }{
+ {
+ name: "empty",
+ args: args{cm: []types.Message{}},
+ want: []testutil.IterVal[slack.Message, error]{},
+ },
+ {
+ name: "one",
+ args: args{cm: []types.Message{
+ {Message: slack.Message{Msg: slack.Msg{Text: "one"}}},
+ }},
+ want: []testutil.IterVal[slack.Message, error]{
+ {T: slack.Message{Msg: slack.Msg{Text: "one"}}, U: nil},
+ },
+ },
+ {
+ name: "two",
+ args: args{cm: []types.Message{
+ {Message: slack.Message{Msg: slack.Msg{Text: "one"}}},
+ {Message: slack.Message{Msg: slack.Msg{Text: "two"}}},
+ }},
+ want: []testutil.IterVal[slack.Message, error]{
+ {T: slack.Message{Msg: slack.Msg{Text: "one"}}, U: nil},
+ {T: slack.Message{Msg: slack.Msg{Text: "two"}}, U: nil},
+ },
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ it := convertMessages(tt.args.cm)
+ var i int
+ for m, err := range it {
+ assert.Equal(t, tt.want[i].T, m)
+ assert.Equal(t, tt.want[i].U, err)
+ i++
+ }
+ })
+ }
+}
diff --git a/internal/testutil/iter.go b/internal/testutil/iter.go
new file mode 100644
index 00000000..144fef90
--- /dev/null
+++ b/internal/testutil/iter.go
@@ -0,0 +1,15 @@
+package testutil
+
+import "iter"
+
+type IterVal[T, U any] struct {
+ T T
+ U U
+}
+
+func Seq2Collect[T, U any](it iter.Seq2[T, U]) (ret []IterVal[T, U]) {
+ for t, u := range it {
+ ret = append(ret, IterVal[T, U]{T: t, U: u})
+ }
+ return
+}
diff --git a/internal/viewer/handlers.go b/internal/viewer/handlers.go
index dfd11f39..66d505ec 100644
--- a/internal/viewer/handlers.go
+++ b/internal/viewer/handlers.go
@@ -4,6 +4,7 @@ import (
"errors"
"fmt"
"io/fs"
+ "iter"
"net/http"
"os"
"path/filepath"
@@ -30,8 +31,8 @@ type mainView struct {
Name string
Type string
Conversation slack.Channel
- Messages []slack.Message
- ThreadMessages []slack.Message
+ Messages iter.Seq2[slack.Message, error]
+ ThreadMessages iter.Seq2[slack.Message, error]
ThreadID string
}
@@ -78,7 +79,7 @@ func maybeReverse(mm []slack.Message) error {
func (v *Viewer) channelHandler(w http.ResponseWriter, r *http.Request, id string) {
ctx := r.Context()
lg := v.lg.With("in", "channelHandler", "channel", id)
- mm, err := v.src.AllMessages(r.Context(), id)
+ it, err := v.src.AllMessages(r.Context(), id)
if err != nil {
if errors.Is(err, fs.ErrNotExist) {
http.NotFound(w, r)
@@ -89,13 +90,7 @@ func (v *Viewer) channelHandler(w http.ResponseWriter, r *http.Request, id strin
return
}
- if err := maybeReverse(mm); err != nil {
- lg.ErrorContext(ctx, "maybeReverse", "error", err)
- http.Error(w, err.Error(), http.StatusInternalServerError)
- return
- }
-
- lg.DebugContext(ctx, "conversation", "id", id, "message_count", len(mm))
+ lg.DebugContext(ctx, "conversation", "id", id)
ci, err := v.src.ChannelInfo(r.Context(), id)
if err != nil {
@@ -106,7 +101,7 @@ func (v *Viewer) channelHandler(w http.ResponseWriter, r *http.Request, id strin
page := v.view()
page.Conversation = *ci
- page.Messages = mm
+ page.Messages = it
template := "index.html" // for deep links
if isHXRequest(r) {
@@ -160,20 +155,21 @@ func (v *Viewer) threadHandler(w http.ResponseWriter, r *http.Request, id string
http.NotFound(w, r)
return
}
+
if strings.HasPrefix(ts, "p") {
ts = structures.ThreadIDtoTS(ts)
}
ctx := r.Context()
lg := v.lg.With("in", "threadHandler", "channel", id, "thread", ts)
- mm, err := v.src.AllThreadMessages(r.Context(), id, ts)
+ itTm, err := v.src.AllThreadMessages(r.Context(), id, ts)
if err != nil {
lg.ErrorContext(ctx, "AllThreadMessages", "error", err)
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
- lg.DebugContext(ctx, "Messages", "mm_count", len(mm))
+ lg.DebugContext(ctx, "Messages")
ci, err := v.src.ChannelInfo(r.Context(), id)
if err != nil {
@@ -184,7 +180,7 @@ func (v *Viewer) threadHandler(w http.ResponseWriter, r *http.Request, id string
page := v.view()
page.Conversation = *ci
- page.ThreadMessages = mm
+ page.ThreadMessages = itTm
page.ThreadID = ts
var template string
@@ -195,17 +191,12 @@ func (v *Viewer) threadHandler(w http.ResponseWriter, r *http.Request, id string
// if we're deep linking, channel view might not contain the messages,
// so we need to fetch them.
- msg, err := v.src.AllMessages(r.Context(), id)
+ itMsg, err := v.src.AllMessages(r.Context(), id)
if err != nil {
lg.ErrorContext(ctx, "AllMessages", "error", err, "template", template)
http.Error(w, err.Error(), http.StatusInternalServerError)
}
- if err := maybeReverse(msg); err != nil {
- lg.ErrorContext(ctx, "maybeReverse", "error", err)
- http.Error(w, err.Error(), http.StatusInternalServerError)
- return
- }
- page.Messages = msg
+ page.Messages = itMsg
}
if err := v.tmpl.ExecuteTemplate(w, template, page); err != nil {
lg.ErrorContext(ctx, "ExecuteTemplate", "error", err, "template", template)
diff --git a/internal/viewer/handlers_test.go b/internal/viewer/handlers_test.go
new file mode 100644
index 00000000..a9584db8
--- /dev/null
+++ b/internal/viewer/handlers_test.go
@@ -0,0 +1 @@
+package viewer
diff --git a/internal/viewer/template.go b/internal/viewer/template.go
index 22b35531..ae6c5bbc 100644
--- a/internal/viewer/template.go
+++ b/internal/viewer/template.go
@@ -23,8 +23,8 @@ func initTemplates(v *Viewer) {
"username": v.username, // username returns the username for the message
"time": localtime,
"rendertext": func(s string) string { return v.r.RenderText(context.Background(), s) }, // render message text
- "render": func(m *slack.Message) template.HTML { return v.r.Render(context.Background(), m) }, // render message
- "is_thread_start": st.IsThreadStart,
+ "render": func(m slack.Message) template.HTML { return v.r.Render(context.Background(), &m) }, // render message
+ "is_thread_start": func(m slack.Message) bool { return st.IsThreadStart(&m) },
},
).ParseFS(fsys, "templates/*.html"))
v.tmpl = tmpl
@@ -47,7 +47,7 @@ const (
sApp
)
-func msgsender(m *slack.Message) sender {
+func msgsender(m slack.Message) sender {
if m.BotID != "" {
if m.Username != "" {
return sApp
@@ -62,7 +62,7 @@ func msgsender(m *slack.Message) sender {
return sUnknown
}
-func (v *Viewer) username(m *slack.Message) string {
+func (v *Viewer) username(m slack.Message) string {
switch msgsender(m) {
case sUser:
return v.um.DisplayName(m.User)
@@ -77,6 +77,6 @@ func (v *Viewer) username(m *slack.Message) string {
}
}
-func isAppMsg(m *slack.Message) bool {
+func isAppMsg(m slack.Message) bool {
return msgsender(m) == sApp
}
diff --git a/internal/viewer/templates/index.html b/internal/viewer/templates/index.html
index e97a4796..d430e825 100644
--- a/internal/viewer/templates/index.html
+++ b/internal/viewer/templates/index.html
@@ -70,15 +70,17 @@ Direct
{{ if .Conversation.ID }}
{{ $id := .Conversation.ID }}
{{ rendername .Conversation }}
-{{ range $i, $el := .Messages }}
+{{ range $el, $err := .Messages }}
- {{ template "render_message" . }}
- {{ if is_thread_start $el }}
-
+ {{ if $err }} Error: {{ $err }}
{{ else }}
+ {{ template "render_message" $el }}
+ {{ if is_thread_start $el }}
+
+ {{ end }}
{{ end }}
{{end}}
@@ -117,9 +119,11 @@ {{ rendername .Conversation }}
Thread: {{ .ThreadID }}
[X]
Link to this thread
-{{ range $i, $el := .ThreadMessages }}
+{{ range $el, $err := .ThreadMessages }}
- {{ template "render_message" $el }}
+ {{ if $err }} Error: {{ $err }}
{{ else }}
+ {{ template "render_message" $el }}
+ {{ end }}
{{end}}
diff --git a/types/message.go b/types/message.go
index bbfc9612..8013810d 100644
--- a/types/message.go
+++ b/types/message.go
@@ -5,6 +5,7 @@ import (
"time"
"github.com/rusq/slack"
+
"github.com/rusq/slackdump/v3/internal/structures"
)
@@ -20,23 +21,23 @@ func (m Message) Datetime() (time.Time, error) {
// IsBotMessage returns true if the message is from a bot.
func (m Message) IsBotMessage() bool {
- return m.Msg.BotID != ""
+ return m.BotID != ""
}
func (m Message) IsThread() bool {
- return m.Msg.ThreadTimestamp != ""
+ return m.ThreadTimestamp != ""
}
// IsThreadParent will return true if the message is the parent message of a
// conversation (has more than 0 replies)
func (m Message) IsThreadParent() bool {
- return m.IsThread() && m.Msg.ReplyCount != 0
+ return m.IsThread() && m.ReplyCount != 0
}
// IsThreadChild will return true if the message is the child message of a
// conversation.
func (m Message) IsThreadChild() bool {
- return m.IsThread() && m.Msg.ReplyCount == 0
+ return m.IsThread() && m.ReplyCount == 0
}
func SortMessages(msgs []Message) {