fix(server): keyset message pagination + attachments in List (F36, F37)

server/internal/api/attachment_migration_test.go

@@ -0,0 +1,76 @@
+package api_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"github.com/Calmingstorm/bastion/server/internal/testutil"
+)
+
+// TestMigration018BackfillsAttachmentPositions verifies that upgrading a live
+// schema with existing multi-attachment messages succeeds, assigns every row a
+// unique zero-based ordinal, and that the down migration removes the column.
+func TestMigration018BackfillsAttachmentPositions(t *testing.T) {
+	dsn := testutil.NewMigrationDB(t)
+	m := newMigrator(t, dsn)
+	if err := m.Migrate(17); err != nil {
+		t.Fatalf("migrate to v17: %v", err)
+	}
+
+	ctx := context.Background()
+	pool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	defer pool.Close()
+
+	mustID := func(query string, args ...any) string {
+		t.Helper()
+		var id string
+		if err := pool.QueryRow(ctx, query, args...).Scan(&id); err != nil {
+			t.Fatalf("seed migration fixture: %v", err)
+		}
+		return id
+	}
+	ownerID := mustID(`INSERT INTO users (username, email, password_hash)
+		VALUES ('owner', 'owner@migration.test', '') RETURNING id`)
+	serverID := mustID(`INSERT INTO servers (name, owner_id) VALUES ('S', $1) RETURNING id`, ownerID)
+	channelID := mustID(`INSERT INTO channels (server_id, name, type)
+		VALUES ($1, 'general', 'text') RETURNING id`, serverID)
+	messageID := mustID(`INSERT INTO messages (channel_id, author_id, content)
+		VALUES ($1, $2, 'attachments') RETURNING id`, channelID, ownerID)
+
+	if _, err := pool.Exec(ctx,
+		`INSERT INTO attachments (message_id, filename, stored_name, content_type, size, url, created_at)
+		 VALUES ($1, 'a.png', 'a', 'image/png', 1, '/a', '2026-01-01T00:00:00Z'),
+		        ($1, 'b.png', 'b', 'image/png', 1, '/b', '2026-01-01T00:00:00Z')`,
+		messageID,
+	); err != nil {
+		t.Fatalf("seed legacy attachments: %v", err)
+	}
+
+	if err := m.Migrate(18); err != nil {
+		t.Fatalf("migrate to v18: %v", err)
+	}
+
+	var count, distinct, minPosition, maxPosition int
+	if err := pool.QueryRow(ctx,
+		`SELECT COUNT(*), COUNT(DISTINCT position), MIN(position), MAX(position)
+		   FROM attachments WHERE message_id = $1`, messageID,
+	).Scan(&count, &distinct, &minPosition, &maxPosition); err != nil {
+		t.Fatalf("read backfilled positions: %v", err)
+	}
+	if count != 2 || distinct != 2 || minPosition != 0 || maxPosition != 1 {
+		t.Fatalf("backfilled positions: count=%d distinct=%d range=%d..%d, want 2/2/0..1",
+			count, distinct, minPosition, maxPosition)
+	}
+
+	if err := m.Migrate(17); err != nil {
+		t.Fatalf("migrate down to v17: %v", err)
+	}
+	if columnExists(t, pool, "attachments", "position") {
+		t.Fatal("position column should be dropped after rollback")
+	}
+}

server/internal/api/messages.go

@@ -353,7 +353,7 @@ func (h *MessageHandler) List(w http.ResponseWriter, r *http.Request) {
 		// Get the created_at of the cursor message
 		var cursorTime time.Time
 		err = h.db.QueryRow(r.Context(),
-			`SELECT created_at FROM messages WHERE id = $1`, beforeID,
+			`SELECT created_at FROM messages WHERE id = $1 AND channel_id = $2`, beforeID, channelID,
 		).Scan(&cursorTime)
 		if errors.Is(err, pgx.ErrNoRows) {
 			writeJSON(w, http.StatusBadRequest, errorResponse("VALIDATION_ERROR", "cursor message not found"))
@@ -364,9 +364,14 @@ func (h *MessageHandler) List(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 
+		// Keyset pagination on (created_at, id): a plain created_at comparison
+		// SKIPS messages that share the cursor's exact timestamp (they are not
+		// strictly older, and were cut from the prior page by LIMIT), and ties
+		// make the order unstable across pages. The row comparison
+		// (a, b) < (c, d) is a < c OR (a = c AND b < d) -- an exact keyset.
 		rows, err = h.db.Query(r.Context(),
-			baseQuery+` AND m.created_at < $2 ORDER BY m.created_at DESC LIMIT $3`,
-			channelID, cursorTime, limit,
+			baseQuery+` AND (m.created_at, m.id) < ($2, $3) ORDER BY m.created_at DESC, m.id DESC LIMIT $4`,
+			channelID, cursorTime, beforeID, limit,
 		)
 		if err != nil {
 			log.Error().Err(err).Msg("failed to list messages")
@@ -375,7 +380,7 @@ func (h *MessageHandler) List(w http.ResponseWriter, r *http.Request) {
 		}
 	} else {
 		rows, err = h.db.Query(r.Context(),
-			baseQuery+` ORDER BY m.created_at DESC LIMIT $2`,
+			baseQuery+` ORDER BY m.created_at DESC, m.id DESC LIMIT $2`,
 			channelID, limit,
 		)
 		if err != nil {
@@ -472,6 +477,45 @@ func (h *MessageHandler) List(w http.ResponseWriter, r *http.Request) {
 				}
 			}
 		}
+
+		// Bulk-fetch attachments (mirrors the reactions fetch above). Without
+		// this, attachments were absent from every List/pagination response and
+		// only arrived via realtime MESSAGE_CREATE -- so any reload or history
+		// scroll showed attachment messages without their attachments.
+		attachmentRows, err := h.db.Query(r.Context(),
+			`SELECT id, message_id, filename, stored_name, content_type, size, url, created_at
+			 FROM attachments WHERE message_id = ANY($1)
+			 ORDER BY message_id ASC, position ASC, id ASC`,
+			messageIDs,
+		)
+		if err != nil {
+			log.Error().Err(err).Msg("failed to list attachments for messages")
+			writeJSON(w, http.StatusInternalServerError, errorResponse("INTERNAL_ERROR", "internal server error"))
+			return
+		}
+		defer attachmentRows.Close()
+
+		attachmentMap := make(map[uuid.UUID][]models.Attachment)
+		for attachmentRows.Next() {
+			var att models.Attachment
+			if err := attachmentRows.Scan(&att.ID, &att.MessageID, &att.Filename, &att.StoredName,
+				&att.ContentType, &att.Size, &att.URL, &att.CreatedAt); err != nil {
+				log.Error().Err(err).Msg("failed to scan attachment for message list")
+				writeJSON(w, http.StatusInternalServerError, errorResponse("INTERNAL_ERROR", "internal server error"))
+				return
+			}
+			attachmentMap[att.MessageID] = append(attachmentMap[att.MessageID], att)
+		}
+		if err := attachmentRows.Err(); err != nil {
+			log.Error().Err(err).Msg("failed to read attachments for message list")
+			writeJSON(w, http.StatusInternalServerError, errorResponse("INTERNAL_ERROR", "internal server error"))
+			return
+		}
+		for i := range messages {
+			if attachments, ok := attachmentMap[messages[i].ID]; ok {
+				messages[i].Attachments = attachments
+			}
+		}
 	}
 
 	writeJSON(w, http.StatusOK, messages)