spark-server/internal/transport/http/inbox_test.go

package httpx

import (
	"encoding/json"
	"fmt"
	"net/http"
	"testing"
	"time"

	"go.uber.org/zap"

	"spark-server/internal/infrastructure/id"
	authlogic "spark-server/internal/module/auth/logic"
	convlogic "spark-server/internal/module/conversation/logic"
	convmodel "spark-server/internal/module/conversation/model"
	deliverlogic "spark-server/internal/module/deliver/logic"
	delivermodel "spark-server/internal/module/deliver/model"
	friendlogic "spark-server/internal/module/friend/logic"
	friendmodel "spark-server/internal/module/friend/model"
	inboxmodel "spark-server/internal/module/inbox/model"
	msglogic "spark-server/internal/module/message/logic"
	msgmodel "spark-server/internal/module/message/model"
	synclogic "spark-server/internal/module/sync/logic"
	syncmodel "spark-server/internal/module/sync/model"
	userlogic "spark-server/internal/module/user/logic"
	usermodel "spark-server/internal/module/user/model"
)

func TestInboxIncrementalHTTP(t *testing.T) {
	ids, err := id.New(1)
	if err != nil {
		t.Fatal(err)
	}
	j, err := authlogic.NewJWT(&authlogic.Options{Secret: "test-secret-test-secret", Issuer: "spark-im"})
	if err != nil {
		t.Fatal(err)
	}
	users := userlogic.NewService(usermodel.NewMemoryRepo(), usermodel.NewMemoryCodes(), ids, nil)
	chats := convmodel.NewMemoryP2P()
	p2p := convlogic.NewP2PService(chats, ids)
	q := delivermodel.NewMemoryQueue()
	store := msgmodel.NewMemoryRepo()
	friends := friendlogic.NewService(
		friendmodel.NewMemoryRepo(), users, ids,
		friendlogic.NewGreeter(p2p, store, ids, q),
	)
	msgs := msglogic.NewService(store, chats, p2p, friends, nil, ids, q)
	inbox := inboxmodel.NewMemoryStore()
	box := syncmodel.NewMemoryStore(ids)
	syncSvc := synclogic.NewService(inbox, box)
	h := NewEngine(zap.NewNop(), &EngineDeps{
		Auth: j, Login: authlogic.NewLoginService(j, users), Users: users, Friends: friends, Messages: msgs, Sync: syncSvc,
	})
	a := loginUser(t, h, "test01@spark.com", 1)
	b := loginUser(t, h, "test02@spark.com", 2)
	apply := authJSON(t, h, &a, http.MethodPost, "/v1/friends/requests", `{"email":"test02@spark.com"}`)
	if apply.Code != "0" {
		t.Fatalf("apply %+v", apply)
	}
	var applied friendlogic.ApplyResult
	if err := json.Unmarshal(apply.Data, &applied); err != nil {
		t.Fatal(err)
	}
	acc := authJSON(t, h, &b, http.MethodPost, "/v1/friends", fmt.Sprintf(`{"request_id":%d}`, applied.RequestID))
	if acc.Code != "0" {
		t.Fatalf("accept %+v", acc)
	}
	sent := authJSON(t, h, &a, http.MethodPost, "/v1/messages",
		fmt.Sprintf(`{"peer_id":%d,"client_msg_id":"c1","content":{"t":1,"text":{"body":"hi"}}}`, b.UserID))
	if sent.Code != "0" {
		t.Fatalf("send %+v", sent)
	}
	var ack msglogic.SendResponse
	if err := json.Unmarshal(sent.Data, &ack); err != nil {
		t.Fatal(err)
	}
	fanout := deliverlogic.InboxFanout(nil, inbox, box)
	drainDeliver(t, q, fanout)
	list := authJSON(t, h, &b, http.MethodGet, "/v1/conversations?dialog_score=0&mailbox_seq=0", "")
	if list.Code != "0" {
		t.Fatalf("inbox %+v", list)
	}
	var delta synclogic.InboxResponse
	if err := json.Unmarshal(list.Data, &delta); err != nil || len(delta.Dialogs) == 0 {
		t.Fatalf("delta %+v %v", delta, err)
	}
	hist := authJSON(t, h, &b, http.MethodGet, fmt.Sprintf("/v1/messages?conversation_id=%d&last_msg_id=0", ack.ConversationID), "")
	if hist.Code != "0" {
		t.Fatalf("history %+v", hist)
	}
	var page msglogic.HistoryResponse
	if err := json.Unmarshal(hist.Data, &page); err != nil || len(page.Items) == 0 {
		t.Fatalf("history data %+v %v", page, err)
	}
	lastID := page.Items[len(page.Items)-1].MsgID
	caughtHist := authJSON(t, h, &b, http.MethodGet, fmt.Sprintf("/v1/messages?conversation_id=%d&last_msg_id=%d", ack.ConversationID, lastID), "")
	if caughtHist.Code != "0" {
		t.Fatalf("caught hist %+v", caughtHist)
	}
	var emptyHist msglogic.HistoryResponse
	if err := json.Unmarshal(caughtHist.Data, &emptyHist); err != nil || len(emptyHist.Items) != 0 {
		t.Fatalf("want no more messages %+v %v", emptyHist, err)
	}
	caught := authJSON(t, h, &b, http.MethodGet, fmt.Sprintf("/v1/conversations?dialog_score=%d&mailbox_seq=%d", delta.DialogScore, delta.MailboxSeq), "")
	if caught.Code != "0" {
		t.Fatalf("caught %+v", caught)
	}
	var empty synclogic.InboxResponse
	if err := json.Unmarshal(caught.Data, &empty); err != nil || len(empty.Dialogs) != 0 {
		t.Fatalf("want empty delta %+v %v", empty, err)
	}
}

func drainDeliver(t *testing.T, q *delivermodel.MemoryQueue, handle deliverlogic.Handler) {
	t.Helper()
	ctx := t.Context()
	for i := 0; i < 16; i++ {
		job, err := q.Take(ctx, 20*time.Millisecond)
		if err != nil {
			t.Fatal(err)
		}
		if job == nil {
			return
		}
		if err := handle(ctx, job); err != nil {
			t.Fatal(err)
		}
	}
}