Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
5.6 kB · 199 lines
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200package atproto
import ( "context" "fmt" "slices" "strings" "sync" "testing" "time"
"github.com/bluesky-social/indigo/util" "github.com/cenkalti/backoff" glex "github.com/streamplace/glex/runtime" "github.com/stretchr/testify/require" "stream.place/streamplace/pkg/bus" "stream.place/streamplace/pkg/comatproto" "stream.place/streamplace/pkg/config" "stream.place/streamplace/pkg/devenv" "stream.place/streamplace/pkg/model" "stream.place/streamplace/pkg/placestream" "stream.place/streamplace/pkg/statedb")
func TestChatMessage(t *testing.T) { dev := devenv.WithDevEnv(t) t.Logf("dev: %+v", dev) cli := config.CLI{ BroadcasterHost: "example.com", DBURL: ":memory:", RelayHost: strings.ReplaceAll(dev.PDSURL, "http://", "ws://"), PLCURL: dev.PLCURL, } t.Logf("cli: %+v", cli) b := bus.NewBus() cli.DataDir = t.TempDir() mod, err := model.MakeDB(":memory:") require.NoError(t, err) state, err := statedb.MakeDB(context.Background(), &cli, nil, mod) require.NoError(t, err) atsync := &ATProtoSynchronizer{ CLI: &cli, StatefulDB: state, Model: mod, Bus: b, }
ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
go func() { err := atsync.StartFirehose(ctx) require.NoError(t, err) close(done) }()
user := dev.CreateAccount(t) user2 := dev.CreateAccount(t)
ch := b.Subscribe(user.DID) defer b.Unsubscribe(user.DID, ch)
// busMessages is appended by the collector goroutine and read by the test // body, so every access goes through busMu -- this test runs under -race. var busMu sync.Mutex busMessages := []bus.Message{} snapshotBus := func() []bus.Message { busMu.Lock() defer busMu.Unlock() return append([]bus.Message(nil), busMessages...) } go func() { for msg := range ch { t.Logf("message: %+v", msg) busMu.Lock() busMessages = append(busMessages, msg) busMu.Unlock() } }()
msg := placestream.ChatMessage{ LexiconTypeID: "place.stream.chat.message", Text: "Hello, world!", CreatedAt: time.Now().Add(-time.Second).Format(util.ISO8601), Streamer: user.DID, }
rec1, err := comatproto.RepoCreateRecord(ctx, user.XRPC, &comatproto.RepoCreateRecord_Input{ Collection: "place.stream.chat.message", Repo: user.DID, Record: &glex.LexiconTypeDecoder{Val: &msg}, }) require.NoError(t, err)
msg2 := placestream.ChatMessage{ LexiconTypeID: "place.stream.chat.message", Text: "Hello, world 2!", CreatedAt: time.Now().Format(util.ISO8601), Streamer: user.DID, }
_, err = comatproto.RepoCreateRecord(ctx, user2.XRPC, &comatproto.RepoCreateRecord_Input{ Collection: "place.stream.chat.message", Repo: user2.DID, Record: &glex.LexiconTypeDecoder{Val: &msg2}, }) require.NoError(t, err)
messages := []placestream.ChatDefs_MessageView{} err = untilNoErrors(t, func() error { messages, err = mod.MostRecentChatMessages(user.DID) if err != nil { return err } if len(messages) != 2 { return fmt.Errorf("expected 2 messages, got %d", len(messages)) } if n := len(snapshotBus()); n != 2 { return fmt.Errorf("expected 2 bus messages, got %d", n) } return nil }) // Reverse the messages slice to match expected order (most recent first) slices.SortFunc(messages, func(a, b placestream.ChatDefs_MessageView) int { aTime := a.Record.Val.(*placestream.ChatMessage).CreatedAt bTime := b.Record.Val.(*placestream.ChatMessage).CreatedAt if aTime < bTime { return -1 } else if aTime > bTime { return 1 } return 0 }) busSnapshot := snapshotBus() slices.SortFunc(busSnapshot, func(a, b bus.Message) int { aTime := a.(*placestream.ChatDefs_MessageView).Record.Val.(*placestream.ChatMessage).CreatedAt bTime := b.(*placestream.ChatDefs_MessageView).Record.Val.(*placestream.ChatMessage).CreatedAt if aTime < bTime { return -1 } else if aTime > bTime { return 1 } return 0 }) require.Equal(t, msg.Text, messages[0].Record.Val.(*placestream.ChatMessage).Text) require.Equal(t, msg2.Text, messages[1].Record.Val.(*placestream.ChatMessage).Text) busMessage1 := busSnapshot[0].(*placestream.ChatDefs_MessageView) busMessage2 := busSnapshot[1].(*placestream.ChatDefs_MessageView) require.Equal(t, msg.Text, busMessage1.Record.Val.(*placestream.ChatMessage).Text) require.Equal(t, msg2.Text, busMessage2.Record.Val.(*placestream.ChatMessage).Text)
rkey := strings.TrimPrefix(rec1.Uri, fmt.Sprintf("at://%s/place.stream.chat.message/", user.DID))
_, err = comatproto.RepoDeleteRecord(ctx, user.XRPC, &comatproto.RepoDeleteRecord_Input{ Collection: "place.stream.chat.message", Repo: user.DID, Rkey: rkey, })
require.NoError(t, err)
err = untilNoErrors(t, func() error { messages, err = mod.MostRecentChatMessages(user.DID) if err != nil { return err } if len(messages) != 1 { return fmt.Errorf("expected 1 message, got %d", len(messages)) } if n := len(snapshotBus()); n != 3 { return fmt.Errorf("expected 3 bus messages, got %d", n) } return nil }) require.NoError(t, err) require.Equal(t, msg2.Text, messages[0].Record.Val.(*placestream.ChatMessage).Text) busMessage3 := snapshotBus()[2].(*placestream.ChatDefs_MessageView) require.Equal(t, true, *busMessage3.Deleted)
cancel() <-done}
func untilNoErrors(t *testing.T, f func() error) error { ticker := backoff.NewTicker(devenv.NewExponentialBackOff()) defer ticker.Stop() var err error for i := 0; i < 10; i++ { err = f() if err == nil { return err } if i < 9 { <-ticker.C } } return err}