260 lines
9.2 KiB
Go
260 lines
9.2 KiB
Go
package app
|
|
|
|
import (
|
|
"bufio"
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"net/url"
|
|
"os"
|
|
"strings"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/alicebob/miniredis/v2"
|
|
miniredisserver "github.com/alicebob/miniredis/v2/server"
|
|
"github.com/gorilla/websocket"
|
|
"github.com/stretchr/testify/require"
|
|
"gorm.io/driver/postgres"
|
|
"gorm.io/gorm"
|
|
"gorm.io/gorm/logger"
|
|
|
|
handlerws "github.com/gochat/gochat/internal/handler/ws"
|
|
"github.com/gochat/gochat/internal/model"
|
|
wspkg "github.com/gochat/gochat/internal/ws"
|
|
)
|
|
|
|
func TestBootstrapRoutesDurableRealtimeToHubAndSSEWithoutDuplicates(t *testing.T) {
|
|
if os.Getenv("GOCHAT_TEST_DB") == "sqlite" {
|
|
t.Skip("bootstrap production wiring requires PostgreSQL")
|
|
}
|
|
dsn := os.Getenv("GOCHAT_TEST_DB_URL")
|
|
if dsn == "" {
|
|
t.Skip("GOCHAT_TEST_DB_URL is not set")
|
|
}
|
|
|
|
db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{Logger: logger.Default.LogMode(logger.Silent)})
|
|
require.NoError(t, err)
|
|
sqlDB, err := db.DB()
|
|
require.NoError(t, err)
|
|
require.NoError(t, sqlDB.Ping())
|
|
t.Cleanup(func() { require.NoError(t, sqlDB.Close()) })
|
|
schema := fmt.Sprintf("hh564_bootstrap_%d", time.Now().UnixNano())
|
|
require.NoError(t, db.Exec(fmt.Sprintf(`CREATE SCHEMA %q`, schema)).Error)
|
|
t.Cleanup(func() { require.NoError(t, db.Exec(fmt.Sprintf(`DROP SCHEMA %q CASCADE`, schema)).Error) })
|
|
databaseURL, err := url.Parse(dsn)
|
|
require.NoError(t, err)
|
|
query := databaseURL.Query()
|
|
query.Set("search_path", schema+",public")
|
|
databaseURL.RawQuery = query.Encode()
|
|
|
|
mini := miniredis.RunT(t)
|
|
t.Setenv("GOCHAT_DATABASE_DSN", databaseURL.String())
|
|
t.Setenv("GOCHAT_DATABASE_RUN_MIGRATIONS", "true")
|
|
t.Setenv("GOCHAT_REDIS_DSN", "redis://"+mini.Addr()+"/0")
|
|
t.Setenv("GOCHAT_JWT_ALLOW_INSECURE_HEADER_AUTH", "true")
|
|
t.Setenv("GOCHAT_SEARCH_ENGINE", "db")
|
|
t.Setenv("GOCHAT_RATE_LIMIT_ENABLED", "false")
|
|
t.Setenv("GOCHAT_LOG_LEVEL", "error")
|
|
t.Chdir("../..")
|
|
|
|
application, err := Bootstrap("default")
|
|
require.NoError(t, err)
|
|
relayCtx, stopRelay := context.WithCancel(context.Background())
|
|
require.NoError(t, application.wsRelay.Start(relayCtx))
|
|
t.Cleanup(func() {
|
|
stopRelay()
|
|
if err := application.Shutdown(2 * time.Second); err != nil && !strings.Contains(err.Error(), "redis: client is closed") {
|
|
t.Errorf("shutdown bootstrap app: %v", err)
|
|
}
|
|
})
|
|
|
|
require.NoError(t, application.db.Exec("DELETE FROM background_jobs").Error)
|
|
stamp := time.Now().UnixNano()
|
|
account := model.Account{Name: fmt.Sprintf("bootstrap-sse-%d", stamp), Active: true}
|
|
require.NoError(t, application.db.Create(&account).Error)
|
|
user := model.User{AccountID: account.ID, Name: "Bootstrap SSE", Email: fmt.Sprintf("bootstrap-sse-%d@example.test", stamp), Password: "unused", Provider: "email", Active: true}
|
|
require.NoError(t, application.db.Create(&user).Error)
|
|
require.NoError(t, application.db.Create(&model.AccountUser{UserID: user.ID, AccountID: account.ID, Role: "administrator"}).Error)
|
|
visitorToken := fmt.Sprintf("bootstrap-visitor-%d", stamp)
|
|
inbox := model.Inbox{AccountID: account.ID, Name: "Bootstrap widget", ChannelType: "Channel::WebWidget", Enabled: true}
|
|
require.NoError(t, application.db.Create(&inbox).Error)
|
|
contact := model.Contact{AccountID: account.ID, Name: "Bootstrap visitor"}
|
|
require.NoError(t, application.db.Create(&contact).Error)
|
|
require.NoError(t, application.db.Create(&model.ContactInbox{ContactID: contact.ID, InboxID: inbox.ID, PubsubToken: visitorToken}).Error)
|
|
|
|
dashboard := handlerws.NewClient(user.ID, account.ID, nil, application.wsHub)
|
|
dashboard.Identifier = fmt.Sprintf(`{"channel":"AccountChannel","account_id":%d}`, account.ID)
|
|
application.wsHub.Register(dashboard)
|
|
visitor := handlerws.NewClient(0, account.ID, nil, application.wsHub)
|
|
visitor.IsContact = true
|
|
visitor.PubsubToken = visitorToken
|
|
visitor.Identifier = fmt.Sprintf(`{"channel":"RoomChannel","pubsub_token":%q}`, visitorToken)
|
|
application.wsHub.Register(visitor)
|
|
t.Cleanup(func() {
|
|
application.wsHub.Unregister(dashboard)
|
|
application.wsHub.Unregister(visitor)
|
|
})
|
|
|
|
server := httptest.NewServer(application.Handler())
|
|
t.Cleanup(server.Close)
|
|
visitorConn, _, err := websocket.DefaultDialer.Dial(
|
|
"ws"+strings.TrimPrefix(server.URL, "http")+"/cable?pubsub_token="+url.QueryEscape(visitorToken),
|
|
nil,
|
|
)
|
|
require.NoError(t, err)
|
|
t.Cleanup(func() { require.NoError(t, visitorConn.Close()) })
|
|
require.NoError(t, visitorConn.SetReadDeadline(time.Now().Add(time.Second)))
|
|
_, _, err = visitorConn.ReadMessage()
|
|
require.NoError(t, err)
|
|
identifier := fmt.Sprintf(`{"channel":"RoomChannel","pubsub_token":%q}`, visitorToken)
|
|
require.NoError(t, visitorConn.WriteJSON(map[string]any{"command": "subscribe", "identifier": identifier}))
|
|
var confirmation map[string]any
|
|
require.NoError(t, visitorConn.ReadJSON(&confirmation))
|
|
require.Equal(t, "confirm_subscription", confirmation["type"])
|
|
require.JSONEq(t, identifier, confirmation["identifier"].(string))
|
|
|
|
requestCtx, cancelRequest := context.WithCancel(context.Background())
|
|
t.Cleanup(cancelRequest)
|
|
req, err := http.NewRequestWithContext(requestCtx, http.MethodGet, fmt.Sprintf("%s/api/v1/accounts/%d/events", server.URL, account.ID), nil)
|
|
require.NoError(t, err)
|
|
req.Header.Set("X-User-ID", fmt.Sprint(user.ID))
|
|
req.Header.Set("X-Account-ID", fmt.Sprint(account.ID))
|
|
resp, err := server.Client().Do(req)
|
|
require.NoError(t, err)
|
|
t.Cleanup(func() { require.NoError(t, resp.Body.Close()) })
|
|
require.Equal(t, http.StatusOK, resp.StatusCode)
|
|
require.Equal(t, "text/event-stream", resp.Header.Get("Content-Type"))
|
|
sse := bufio.NewReader(resp.Body)
|
|
eventType, _ := readBootstrapSSE(t, sse)
|
|
require.Equal(t, "connected", eventType)
|
|
|
|
tokenChannel := wspkg.RedisPrefixRoom + "pubsub_token_" + visitorToken
|
|
var tokenFailed atomic.Bool
|
|
mini.Server().SetPreHook(func(peer *miniredisserver.Peer, command string, args ...string) bool {
|
|
if strings.EqualFold(command, "publish") && len(args) > 0 && args[0] == tokenChannel && tokenFailed.CompareAndSwap(false, true) {
|
|
peer.WriteError("token room unavailable")
|
|
return true
|
|
}
|
|
return false
|
|
})
|
|
|
|
require.NoError(t, application.eventPublisher.PublishWidgetEvent(account.ID, visitorToken, wspkg.EventMessageCreated, map[string]any{"id": 7}))
|
|
processed, err := application.workerPool.ProcessOne(context.Background())
|
|
require.True(t, processed)
|
|
require.NoError(t, err)
|
|
requireHubEvent(t, dashboard.Send)
|
|
eventType, payload := readBootstrapSSE(t, sse)
|
|
require.Equal(t, wspkg.EventMessageCreated, eventType)
|
|
require.JSONEq(t, `{"id":7}`, payload)
|
|
|
|
processed, err = application.workerPool.ProcessOne(context.Background())
|
|
require.True(t, processed)
|
|
require.ErrorContains(t, err, "token room unavailable")
|
|
select {
|
|
case <-visitor.Send:
|
|
t.Fatal("token subscriber received failed delivery")
|
|
default:
|
|
}
|
|
require.NoError(t, application.db.Model(&model.BackgroundJob{}).
|
|
Where("status = ?", model.BackgroundJobStatusRetrying).
|
|
Update("scheduled_at", time.Now().Add(-time.Second)).Error)
|
|
|
|
processed, err = application.workerPool.ProcessOne(context.Background())
|
|
require.True(t, processed)
|
|
require.NoError(t, err)
|
|
requireHubEvent(t, visitor.Send)
|
|
var delivered struct {
|
|
Identifier string `json:"identifier"`
|
|
Message json.RawMessage `json:"message"`
|
|
}
|
|
require.NoError(t, visitorConn.ReadJSON(&delivered))
|
|
require.JSONEq(t, identifier, delivered.Identifier)
|
|
var widgetEvent wspkg.WSMessage
|
|
require.NoError(t, json.Unmarshal(delivered.Message, &widgetEvent))
|
|
require.Equal(t, wspkg.EventMessageCreated, widgetEvent.Event)
|
|
require.Equal(t, float64(7), widgetEvent.Data.(map[string]any)["id"])
|
|
require.True(t, tokenFailed.Load())
|
|
assertNoHubEvent(t, dashboard.Send)
|
|
assertNoHubEvent(t, visitor.Send)
|
|
assertNoBootstrapSSE(t, sse)
|
|
}
|
|
|
|
func readBootstrapSSE(t *testing.T, reader *bufio.Reader) (string, string) {
|
|
t.Helper()
|
|
resultCh := make(chan bootstrapSSEResult, 1)
|
|
go func() {
|
|
resultCh <- scanBootstrapSSE(reader)
|
|
}()
|
|
select {
|
|
case got := <-resultCh:
|
|
require.NoError(t, got.err)
|
|
return got.event, got.data
|
|
case <-time.After(time.Second):
|
|
t.Fatal("timed out waiting for SSE event")
|
|
return "", ""
|
|
}
|
|
}
|
|
|
|
func requireHubEvent(t *testing.T, events <-chan []byte) {
|
|
t.Helper()
|
|
select {
|
|
case <-events:
|
|
case <-time.After(time.Second):
|
|
t.Fatal("timed out waiting for Hub event")
|
|
}
|
|
}
|
|
|
|
func assertNoHubEvent(t *testing.T, events <-chan []byte) {
|
|
t.Helper()
|
|
time.Sleep(20 * time.Millisecond)
|
|
select {
|
|
case <-events:
|
|
t.Fatal("subscriber received duplicate event")
|
|
default:
|
|
}
|
|
}
|
|
|
|
func assertNoBootstrapSSE(t *testing.T, reader *bufio.Reader) {
|
|
t.Helper()
|
|
eventCh := make(chan bootstrapSSEResult, 1)
|
|
go func() {
|
|
eventCh <- scanBootstrapSSE(reader)
|
|
}()
|
|
select {
|
|
case event := <-eventCh:
|
|
require.NoError(t, event.err)
|
|
t.Fatalf("SSE subscriber received duplicate %s", event.event)
|
|
case <-time.After(50 * time.Millisecond):
|
|
}
|
|
}
|
|
|
|
type bootstrapSSEResult struct {
|
|
event string
|
|
data string
|
|
err error
|
|
}
|
|
|
|
func scanBootstrapSSE(reader *bufio.Reader) bootstrapSSEResult {
|
|
var got bootstrapSSEResult
|
|
for {
|
|
line, err := reader.ReadString('\n')
|
|
if err != nil {
|
|
got.err = err
|
|
return got
|
|
}
|
|
line = strings.TrimSpace(line)
|
|
switch {
|
|
case strings.HasPrefix(line, "event: "):
|
|
got.event = strings.TrimPrefix(line, "event: ")
|
|
case strings.HasPrefix(line, "data: "):
|
|
got.data = strings.TrimPrefix(line, "data: ")
|
|
case line == "":
|
|
return got
|
|
}
|
|
}
|
|
}
|