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" "spark-server/internal/module/user/logictest" ) 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, _ := logictest.UsersAndSMS(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) } } }