refactor: derive hasMore from nextPageToken instead of separate parameter

Remove hasMore as an explicit parameter from IListMessagesProgress.OnComplete().
The hasMore field is now automatically derived from whether nextPageToken is empty.

This simplifies the interface by removing redundant state - the information about
whether more pages exist is already encoded in the presence/absence of nextPageToken.

Changes:
- Remove hasMore parameter from OnComplete() interface (4 params -> 3 params)
- Derive HasMore automatically: HasMore = nextPageToken != ""
- Update all call sites to pass 3 parameters instead of 4
- Update integration tests to check nextPageToken instead of hasMore
- Update mocks to match new 3-parameter signature

The protobuf has_more field is still populated for wire compatibility, but is no
longer passed as a separate parameter through the application layers.

Author: c-julin

backend/pkg/api/connect/service/console/stream_progress_reporter.go

@@ -216,7 +216,7 @@ func (p *streamProgressReporter) OnMessage(message *console.TopicMessage) {
 	}
 }
 
-func (p *streamProgressReporter) OnComplete(elapsedMs int64, isCancelled bool, nextPageToken string, hasMore bool) {
+func (p *streamProgressReporter) OnComplete(elapsedMs int64, isCancelled bool, nextPageToken string) {
 	p.writeMutex.Lock()
 	defer p.writeMutex.Unlock()
 
@@ -226,7 +226,7 @@ func (p *streamProgressReporter) OnComplete(elapsedMs int64, isCancelled bool, n
 		MessagesConsumed: p.messagesConsumed.Load(),
 		BytesConsumed:    p.bytesConsumed.Load(),
 		NextPageToken:    nextPageToken,
-		HasMore:          hasMore,
+		HasMore:          nextPageToken != "",
 	}
 
 	if err := p.stream.Send(

backend/pkg/api/handle_topic_messages_integration_test.go

@@ -1145,7 +1145,6 @@ func (s *APIIntegrationTestSuite) TestListMessages_Pagination() {
 		// Verify first page results
 		assert.Equal(int64(50), done.MessagesConsumed, "Should consume 50 messages (default page size)")
 		assert.NotEmpty(done.NextPageToken, "Should have next page token")
-		assert.True(done.HasMore, "Should have more messages")
 		assert.Len(messages, 50, "Should return 50 messages")
 
 		// Verify descending order within partitions
@@ -1202,9 +1201,9 @@ func (s *APIIntegrationTestSuite) TestListMessages_Pagination() {
 			allMessages = append(allMessages, pageMessages...)
 			pageCount++
 
-			t.Logf("Page %d: fetched %d messages, hasMore=%v", pageCount, len(pageMessages), done.HasMore)
+			t.Logf("Page %d: fetched %d messages, nextPageToken=%s", pageCount, len(pageMessages), done.NextPageToken)
 
-			if !done.HasMore || done.NextPageToken == "" {
+			if done.NextPageToken == "" {
 				break
 			}
 
@@ -1269,7 +1268,6 @@ func (s *APIIntegrationTestSuite) TestListMessages_Pagination() {
 
 		// In legacy mode, should not have pagination tokens
 		assert.Empty(done.NextPageToken, "Legacy mode should not return page token")
-		assert.False(done.HasMore, "Legacy mode should not set hasMore")
 		assert.LessOrEqual(len(messages), 50, "Should respect maxResults limit")
 	})
 }

backend/pkg/console/list_messages.go

@@ -81,7 +81,7 @@ type IListMessagesProgress interface {
 	OnPhase(name string) // todo(?): eventually we might want to convert this into an enum
 	OnMessage(message *TopicMessage)
 	OnMessageConsumed(size int64)
-	OnComplete(elapsedMs int64, isCancelled bool, nextPageToken string, hasMore bool)
+	OnComplete(elapsedMs int64, isCancelled bool, nextPageToken string)
 	OnError(msg string)
 }
 
@@ -224,7 +224,6 @@ func (s *Service) ListMessages(ctx context.Context, listReq ListMessageRequest,
 	// Use pagination mode if MessageCount == -1
 	var consumeRequests map[int32]*PartitionConsumeRequest
 	var nextPageToken string
-	var hasMore bool
 	var token *PageToken // Declare token outside if block so it's accessible later
 
 	if listReq.MessageCount == -1 {
@@ -260,7 +259,7 @@ func (s *Service) ListMessages(ctx context.Context, listReq ListMessageRequest,
 			}
 		}
 
-		consumeRequests, nextPageToken, hasMore, err = s.calculateConsumeRequestsWithPageToken(ctx, token, partitionIDs, startOffsets, endOffsets)
+		consumeRequests, nextPageToken, _, err = s.calculateConsumeRequestsWithPageToken(ctx, token, partitionIDs, startOffsets, endOffsets)
 		if err != nil {
 			return fmt.Errorf("failed to calculate consume request with page token: %w", err)
 		}
@@ -274,7 +273,7 @@ func (s *Service) ListMessages(ctx context.Context, listReq ListMessageRequest,
 
 	if len(consumeRequests) == 0 {
 		// No partitions/messages to consume, we can quit early.
-		progress.OnComplete(time.Since(start).Milliseconds(), false, "", false)
+		progress.OnComplete(time.Since(start).Milliseconds(), false, "")
 		return nil
 	}
 	// Determine direction based on mode
@@ -305,7 +304,7 @@ func (s *Service) ListMessages(ctx context.Context, listReq ListMessageRequest,
 	}
 
 	isCancelled := ctx.Err() != nil
-	progress.OnComplete(time.Since(start).Milliseconds(), isCancelled, nextPageToken, hasMore)
+	progress.OnComplete(time.Since(start).Milliseconds(), isCancelled, nextPageToken)
 	if isCancelled {
 		return errors.New("request was cancelled while waiting for messages")
 	}