package repository import ( "context" "encoding/json" "errors" "fmt" "os" "path/filepath" "testing" "github.com/google/uuid" "github.com/joho/godotenv" "github.com/prabhupadasays/spsays-bff/internal/model" "go.uber.org/zap" ) // These exercise real SQL against a real Postgres — the LATERAL joins, the // FOR UPDATE sequencing and the partial indexes are the parts most likely to be // wrong, and none of them can be checked without a database. Skipped when no // POSTGRES_URI is configured, so `go test ./...` still passes anywhere. func testRepo(t *testing.T) (*ChatRepo, *Postgres) { t.Helper() uri := os.Getenv("POSTGRES_URI") if uri == "" { _ = godotenv.Load(filepath.Join("..", "..", ".env")) uri = os.Getenv("POSTGRES_URI") } if uri == "" { t.Skip("POSTGRES_URI not set — skipping chat repository integration tests") } pg, err := NewPostgres(context.Background(), uri, zap.NewNop()) if err != nil { t.Skipf("postgres unreachable: %v", err) } t.Cleanup(pg.Close) return NewChatRepo(pg), pg } // makeUser inserts a throwaway user and removes it (and anything cascading // from it) when the test finishes. func makeUser(t *testing.T, pg *Postgres, label string) string { t.Helper() ctx := context.Background() id := uuid.NewString() suffix := id[:8] _, err := pg.Pool.Exec(ctx, ` INSERT INTO users (id, email, password_hash, name, handle) VALUES ($1, $2, 'x', $3, $4)`, id, fmt.Sprintf("chattest-%s@example.invalid", suffix), "Chat Test "+label, "chattest_"+suffix) if err != nil { t.Fatalf("create test user: %v", err) } t.Cleanup(func() { _, _ = pg.Pool.Exec(context.Background(), `DELETE FROM users WHERE id = $1`, id) }) return id } // dropConversation removes a thread when the test ends; messages, participants // and attachments cascade with it. func dropConversation(t *testing.T, pg *Postgres, id string) { t.Cleanup(func() { _, _ = pg.Pool.Exec(context.Background(), `DELETE FROM conversations WHERE id = $1`, id) }) } func send(t *testing.T, r *ChatRepo, convID, sender, body string) *model.Message { t.Helper() m, _, err := r.Append(context.Background(), AppendParams{ ConversationID: convID, SenderID: &sender, Kind: model.MessageText, Body: body, }) if err != nil { t.Fatalf("append %q: %v", body, err) } return m } func TestEnsureDM_IsIdempotentFromEitherSide(t *testing.T) { r, pg := testRepo(t) ctx := context.Background() a, b := makeUser(t, pg, "A"), makeUser(t, pg, "B") first, err := r.EnsureDM(ctx, a, b) if err != nil { t.Fatalf("EnsureDM: %v", err) } dropConversation(t, pg, first.ID) // Same pair, opposite argument order — must resolve to the same thread, // which is the whole point of the sorted dm_key. second, err := r.EnsureDM(ctx, b, a) if err != nil { t.Fatalf("EnsureDM reversed: %v", err) } if second.ID != first.ID { t.Errorf("reversed EnsureDM made a second thread: %s vs %s", second.ID, first.ID) } for _, u := range []string{a, b} { if _, err := r.Participant(ctx, first.ID, u); err != nil { t.Errorf("user %s is not a participant: %v", u, err) } } } func TestParticipant_RejectsOutsiders(t *testing.T) { r, pg := testRepo(t) ctx := context.Background() a, b, outsider := makeUser(t, pg, "A"), makeUser(t, pg, "B"), makeUser(t, pg, "C") c, err := r.EnsureDM(ctx, a, b) if err != nil { t.Fatal(err) } dropConversation(t, pg, c.ID) if _, err := r.Participant(ctx, c.ID, outsider); !errors.Is(err, ErrNotParticipant) { t.Errorf("outsider got %v, want ErrNotParticipant", err) } } func TestAppend_AssignsGaplessSequence(t *testing.T) { r, pg := testRepo(t) ctx := context.Background() a, b := makeUser(t, pg, "A"), makeUser(t, pg, "B") c, err := r.EnsureDM(ctx, a, b) if err != nil { t.Fatal(err) } dropConversation(t, pg, c.ID) for i := int64(1); i <= 3; i++ { m := send(t, r, c.ID, a, fmt.Sprintf("msg %d", i)) if m.Seq != i { t.Errorf("message %d got seq %d", i, m.Seq) } } var lastSeq int64 if err := pg.Pool.QueryRow(ctx, `SELECT last_seq FROM conversations WHERE id=$1`, c.ID).Scan(&lastSeq); err != nil { t.Fatal(err) } if lastSeq != 3 { t.Errorf("conversation.last_seq = %d, want 3", lastSeq) } } // A retried send must return the original message and must NOT burn a sequence // number — a gap would break the client's catch-up cursor. func TestAppend_DeduplicatesRetries(t *testing.T) { r, pg := testRepo(t) ctx := context.Background() a, b := makeUser(t, pg, "A"), makeUser(t, pg, "B") c, err := r.EnsureDM(ctx, a, b) if err != nil { t.Fatal(err) } dropConversation(t, pg, c.ID) clientID := uuid.NewString() p := AppendParams{ConversationID: c.ID, SenderID: &a, Kind: model.MessageText, Body: "only once", ClientMsgID: &clientID} first, created, err := r.Append(ctx, p) if err != nil || !created { t.Fatalf("first append: created=%v err=%v", created, err) } again, created, err := r.Append(ctx, p) if err != nil { t.Fatalf("retry: %v", err) } if created { t.Error("retry reported created=true") } if again.ID != first.ID || again.Seq != first.Seq { t.Errorf("retry returned a different message: %s/%d vs %s/%d", again.ID, again.Seq, first.ID, first.Seq) } var count, lastSeq int64 _ = pg.Pool.QueryRow(ctx, `SELECT count(*), max(seq) FROM messages WHERE conversation_id=$1`, c.ID).Scan(&count, &lastSeq) if count != 1 || lastSeq != 1 { t.Errorf("after retry: %d messages, last seq %d — want 1 and 1", count, lastSeq) } } func TestMessages_KeysetPaging(t *testing.T) { r, pg := testRepo(t) ctx := context.Background() a, b := makeUser(t, pg, "A"), makeUser(t, pg, "B") c, err := r.EnsureDM(ctx, a, b) if err != nil { t.Fatal(err) } dropConversation(t, pg, c.ID) for i := 1; i <= 5; i++ { send(t, r, c.ID, a, fmt.Sprintf("m%d", i)) } page1, err := r.Messages(ctx, c.ID, a, 0, 2) if err != nil { t.Fatal(err) } if len(page1) != 2 || page1[0].Seq != 5 || page1[1].Seq != 4 { t.Fatalf("page1 = %v, want seq 5,4", seqsOf(page1)) } page2, err := r.Messages(ctx, c.ID, a, page1[len(page1)-1].Seq, 2) if err != nil { t.Fatal(err) } if len(page2) != 2 || page2[0].Seq != 3 || page2[1].Seq != 2 { t.Fatalf("page2 = %v, want seq 3,2", seqsOf(page2)) } // entities must come back as a JSON array, never null — the client indexes // into it without a nil check. var spans []map[string]any if err := json.Unmarshal(page1[0].Entities, &spans); err != nil { t.Errorf("entities not valid JSON: %v (%s)", err, page1[0].Entities) } } // Delete-for-me must hide a message from that user's history AND from their // inbox preview, while leaving it visible to everyone else. func TestMessageHides_ArePerUser(t *testing.T) { r, pg := testRepo(t) ctx := context.Background() a, b := makeUser(t, pg, "A"), makeUser(t, pg, "B") c, err := r.EnsureDM(ctx, a, b) if err != nil { t.Fatal(err) } dropConversation(t, pg, c.ID) send(t, r, c.ID, a, "keep") hidden := send(t, r, c.ID, a, "hide me") if _, err := pg.Pool.Exec(ctx, `INSERT INTO message_hides (message_id, user_id) VALUES ($1,$2)`, hidden.ID, a); err != nil { t.Fatal(err) } mine, _ := r.Messages(ctx, c.ID, a, 0, 10) if len(mine) != 1 || mine[0].Body != "keep" { t.Errorf("hider still sees %v", bodiesOf(mine)) } theirs, _ := r.Messages(ctx, c.ID, b, 0, 10) if len(theirs) != 2 { t.Errorf("other participant sees %v, want both", bodiesOf(theirs)) } inbox, err := r.Inbox(ctx, a, []string{model.ConversationDM}, 10) if err != nil { t.Fatal(err) } row := findConv(inbox, c.ID) if row == nil || row.LastMessage == nil { t.Fatal("conversation missing from hider's inbox") } if row.LastMessage.Body != "keep" { t.Errorf("hider's preview = %q, want the un-hidden message", row.LastMessage.Body) } } func TestInbox_UnreadPeerAndPreview(t *testing.T) { r, pg := testRepo(t) ctx := context.Background() a, b := makeUser(t, pg, "A"), makeUser(t, pg, "B") c, err := r.EnsureDM(ctx, a, b) if err != nil { t.Fatal(err) } dropConversation(t, pg, c.ID) send(t, r, c.ID, a, "first") send(t, r, c.ID, a, "latest") // Recipient has read nothing. inbox, err := r.Inbox(ctx, b, []string{model.ConversationDM}, 10) if err != nil { t.Fatal(err) } row := findConv(inbox, c.ID) if row == nil { t.Fatal("thread missing from recipient inbox") } if row.Unread != 2 { t.Errorf("unread = %d, want 2", row.Unread) } if row.LastMessage == nil || row.LastMessage.Body != "latest" { t.Errorf("preview = %+v, want body %q", row.LastMessage, "latest") } if row.LastMessage != nil && row.LastMessage.SenderName == "" { t.Error("preview has no sender name — the inbox line needs it") } if row.Peer == nil || row.Peer.ID != a { t.Errorf("peer = %+v, want user A", row.Peer) } // The sender's own cursor advanced on send, so their unread is zero. senderInbox, _ := r.Inbox(ctx, a, []string{model.ConversationDM}, 10) if sr := findConv(senderInbox, c.ID); sr == nil || sr.Unread != 0 { t.Errorf("sender unread = %v, want 0", sr) } if err := r.MarkRead(ctx, c.ID, b, 2); err != nil { t.Fatal(err) } after, _ := r.Inbox(ctx, b, []string{model.ConversationDM}, 10) if row := findConv(after, c.ID); row == nil || row.Unread != 0 { t.Errorf("after MarkRead unread = %v, want 0", row) } } // Cursors must never move backwards: a replayed or out-of-order receipt would // otherwise resurrect already-read messages as unread. func TestCursors_NeverGoBackwards(t *testing.T) { r, pg := testRepo(t) ctx := context.Background() a, b := makeUser(t, pg, "A"), makeUser(t, pg, "B") c, err := r.EnsureDM(ctx, a, b) if err != nil { t.Fatal(err) } dropConversation(t, pg, c.ID) for i := 0; i < 3; i++ { send(t, r, c.ID, a, "m") } if err := r.MarkRead(ctx, c.ID, b, 3); err != nil { t.Fatal(err) } if err := r.MarkRead(ctx, c.ID, b, 1); err != nil { // stale replay t.Fatal(err) } p, err := r.Participant(ctx, c.ID, b) if err != nil { t.Fatal(err) } if p.LastReadSeq != 3 { t.Errorf("last_read_seq = %d after a stale replay, want 3", p.LastReadSeq) } if p.LastDeliveredSeq != 3 { t.Errorf("reading should imply delivery: last_delivered_seq = %d, want 3", p.LastDeliveredSeq) } } func TestReceipts_ExcludeCallerAndUseMinimum(t *testing.T) { r, pg := testRepo(t) ctx := context.Background() a, b, cUser := makeUser(t, pg, "A"), makeUser(t, pg, "B"), makeUser(t, pg, "C") conv, err := r.CreateGroup(ctx, a, "Test group", []string{b, cUser}) if err != nil { t.Fatal(err) } dropConversation(t, pg, conv.ID) for i := 0; i < 3; i++ { send(t, r, conv.ID, a, "m") } if err := r.MarkDelivered(ctx, conv.ID, b, 3); err != nil { t.Fatal(err) } if err := r.MarkRead(ctx, conv.ID, b, 3); err != nil { t.Fatal(err) } // C has received nothing yet. rc, err := r.Receipts(ctx, conv.ID, a) if err != nil { t.Fatal(err) } if rc.ReadSeq != 0 || rc.DeliveredSeq != 0 { t.Errorf("receipts = %+v, want zeros while one member is behind", rc) } if err := r.MarkRead(ctx, conv.ID, cUser, 3); err != nil { t.Fatal(err) } rc, _ = r.Receipts(ctx, conv.ID, a) if rc.ReadSeq != 3 || rc.DeliveredSeq != 3 { t.Errorf("receipts = %+v, want 3/3 once everyone has read", rc) } } func TestEnsureAIThread_OnePerUserAndSeparateTab(t *testing.T) { r, pg := testRepo(t) ctx := context.Background() a := makeUser(t, pg, "A") first, err := r.EnsureAIThread(ctx, a) if err != nil { t.Fatal(err) } dropConversation(t, pg, first.ID) again, err := r.EnsureAIThread(ctx, a) if err != nil { t.Fatal(err) } if again.ID != first.ID { t.Errorf("second call made another AI thread: %s vs %s", again.ID, first.ID) } // The tabs are a kind filter: an AI thread must not appear under Chats. chats, _ := r.Inbox(ctx, a, []string{model.ConversationDM, model.ConversationGroup}, 10) if findConv(chats, first.ID) != nil { t.Error("AI thread leaked into the Chats tab") } kautilya, _ := r.Inbox(ctx, a, []string{model.ConversationAI}, 10) if findConv(kautilya, first.ID) == nil { t.Error("AI thread missing from the Kautilya tab") } } func findConv(items []model.InboxItem, id string) *model.InboxItem { for i := range items { if items[i].ID == id { return &items[i] } } return nil } func seqsOf(msgs []model.Message) []int64 { out := make([]int64, len(msgs)) for i, m := range msgs { out[i] = m.Seq } return out } func bodiesOf(msgs []model.Message) []string { out := make([]string, len(msgs)) for i, m := range msgs { out[i] = m.Body } return out }