// Licensed to Apache Software Foundation (ASF) under one or more contributor
// license agreements. See the NOTICE file distributed with
// this work for additional information regarding copyright
// ownership. Apache Software Foundation (ASF) licenses this file to you under
// the Apache License, Version 2.0 (the "License"); you may
// not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

package sub

import (
	"errors"
	"fmt"
	"hash/crc32"
	"io"
	"time"

	"github.com/apache/skywalking-banyandb/api/data"
	apiversion "github.com/apache/skywalking-banyandb/api/proto/banyandb"
	clusterv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/cluster/v1"
	"github.com/apache/skywalking-banyandb/banyand/internal/storage"
	"github.com/apache/skywalking-banyandb/banyand/queue"
	"github.com/apache/skywalking-banyandb/pkg/bus"
)

func checkSyncVersionCompatibility(versionInfo *clusterv1.VersionInfo) (*clusterv1.VersionCompatibility, clusterv1.SyncStatus) {
	if versionInfo == nil {
		return &clusterv1.VersionCompatibility{
			Supported:                   true,
			ServerApiVersion:            apiversion.Version,
			SupportedApiVersions:        []string{apiversion.Version},
			ServerFileFormatVersion:     storage.GetCurrentVersion(),
			SupportedFileFormatVersions: storage.GetCompatibleVersions(),
			Reason:                      "No version info provided, assuming compatible",
		}, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_RECEIVED
	}

	serverAPIVersion := apiversion.Version
	serverFileFormatVersion := storage.GetCurrentVersion()
	compatibleFileFormatVersions := storage.GetCompatibleVersions()

	// Check API version compatibility
	apiCompatible := versionInfo.ApiVersion == serverAPIVersion

	// Check file format version compatibility
	fileFormatCompatible := false
	if versionInfo.FileFormatVersion == serverFileFormatVersion {
		fileFormatCompatible = true
	} else {
		// Check if client's file format version is in our compatible list
		for _, compatVer := range compatibleFileFormatVersions {
			if compatVer == versionInfo.FileFormatVersion {
				fileFormatCompatible = true
				break
			}
		}
	}

	versionCompatibility := &clusterv1.VersionCompatibility{
		ServerApiVersion:            serverAPIVersion,
		SupportedApiVersions:        []string{serverAPIVersion},
		ServerFileFormatVersion:     serverFileFormatVersion,
		SupportedFileFormatVersions: compatibleFileFormatVersions,
	}

	switch {
	case !apiCompatible && !fileFormatCompatible:
		versionCompatibility.Supported = false
		versionCompatibility.Reason = fmt.Sprintf("API version %s not supported (server: %s) and file format version %s not compatible (server: %s, supported: %v)",
			versionInfo.ApiVersion, serverAPIVersion, versionInfo.FileFormatVersion, serverFileFormatVersion, compatibleFileFormatVersions)
		return versionCompatibility, clusterv1.SyncStatus_SYNC_STATUS_VERSION_UNSUPPORTED
	case !apiCompatible:
		versionCompatibility.Supported = false
		versionCompatibility.Reason = fmt.Sprintf("API version %s not supported (server: %s)", versionInfo.ApiVersion, serverAPIVersion)
		return versionCompatibility, clusterv1.SyncStatus_SYNC_STATUS_VERSION_UNSUPPORTED
	case !fileFormatCompatible:
		versionCompatibility.Supported = false
		versionCompatibility.Reason = fmt.Sprintf("File format version %s not compatible (server: %s, supported: %v)",
			versionInfo.FileFormatVersion, serverFileFormatVersion, compatibleFileFormatVersions)
		return versionCompatibility, clusterv1.SyncStatus_SYNC_STATUS_FORMAT_VERSION_MISMATCH
	}

	versionCompatibility.Supported = true
	versionCompatibility.Reason = "Client version compatible with server"
	return versionCompatibility, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_RECEIVED
}

type chunkBuffer struct {
	chunks        map[uint32]*clusterv1.SyncPartRequest
	lastActivity  time.Time
	bufferTimeout time.Duration
	expectedIndex uint32
	maxBufferSize uint32
}

type syncSession struct {
	startTime      time.Time
	metadata       *clusterv1.SyncMetadata
	partsProgress  map[int]*partProgress
	partCtx        *queue.ChunkedSyncPartContext
	chunkBuffer    *chunkBuffer
	sessionID      string
	errorMsg       string
	totalReceived  uint64
	chunksReceived uint32
	completed      bool
}

type partProgress struct {
	totalBytes    uint32
	receivedBytes uint32
	completed     bool
}

// SyncPart implements clusterv1.ChunkedSyncServiceServer.
func (s *server) SyncPart(stream clusterv1.ChunkedSyncService_SyncPartServer) error {
	ctx := stream.Context()
	var currentSession *syncSession
	var sessionID string
	defer func() {
		if currentSession == nil {
			return
		}
		if currentSession.partCtx != nil {
			if closeErr := currentSession.partCtx.Close(); closeErr != nil {
				s.log.Error().Err(closeErr).Str("session_id", currentSession.sessionID).Msg("failed to close session partCtx")
			}
		}
	}()

	for {
		select {
		case <-ctx.Done():
			if s.metrics != nil && currentSession != nil && !currentSession.completed {
				op, grp, sn, sr, st := s.resolveSessionLabels(currentSession)
				s.metrics.totalErr.Inc(1, op, grp, sn, sr, st, "ctx_done")
			}
			return ctx.Err()
		default:
		}

		req, recvErr := stream.Recv()
		if errors.Is(recvErr, io.EOF) {
			if s.metrics != nil && currentSession != nil && !currentSession.completed {
				op, grp, sn, sr, st := s.resolveSessionLabels(currentSession)
				s.metrics.totalErr.Inc(1, op, grp, sn, sr, st, "eof")
			}
			break
		}
		if recvErr != nil {
			if s.metrics != nil && currentSession != nil && !currentSession.completed {
				op, grp, sn, sr, st := s.resolveSessionLabels(currentSession)
				s.metrics.totalErr.Inc(1, op, grp, sn, sr, st, "stream_error")
			}
			s.log.Error().Err(recvErr).Msg("failed to receive chunk")
			return recvErr
		}

		sessionID = req.SessionId

		if req.GetMetadata() != nil {
			currentSession = s.startOrSwitchSession(sessionID, req, currentSession)
		}

		if currentSession == nil {
			errMsg := fmt.Sprintf("session %s not found", sessionID)
			s.log.Error().Str("session_id", sessionID).Msg(errMsg)
			return s.sendResponse(stream, req, clusterv1.SyncStatus_SYNC_STATUS_SESSION_NOT_FOUND, errMsg, nil)
		}

		if req.GetCompletion() != nil {
			return s.handleCompletion(stream, currentSession, req)
		}

		if processErr := s.processChunk(stream, currentSession, req); processErr != nil {
			s.log.Error().Err(processErr).Str("session_id", sessionID).Msg("failed to process chunk")
			return processErr
		}
	}

	return nil
}

// resolveSessionLabels returns the label tuple for the current session.
func (s *server) resolveSessionLabels(session *syncSession) (operation, group, senderNode, senderRole, senderTier string) {
	operation = data.OperationFileSyncValue
	if session.metadata != nil {
		group = session.metadata.GetGroup()
		senderNode = session.metadata.GetSenderNode()
		senderRole = session.metadata.GetSenderRole()
		senderTier = session.metadata.GetSenderTier()
	}
	return
}

func (s *server) startOrSwitchSession(sessionID string, req *clusterv1.SyncPartRequest, previousSession *syncSession) *syncSession {
	if previousSession != nil {
		s.cleanupPreviousSession(previousSession)
	}

	newSession := &syncSession{
		sessionID:      sessionID,
		metadata:       req.GetMetadata(),
		startTime:      time.Now(),
		chunksReceived: 0,
		partsProgress:  make(map[int]*partProgress),
	}
	if s.metrics != nil {
		op, grp, sn, sr, st := s.resolveSessionLabels(newSession)
		s.metrics.totalStarted.Inc(1, op, grp, sn, sr, st)
	}
	if dl := s.log.Debug(); dl.Enabled() {
		dl.Str("session_id", sessionID).
			Str("topic", req.GetMetadata().Topic).
			Uint32("total_parts", req.GetMetadata().TotalParts).
			Msg("started chunked sync session")
	}
	return newSession
}

func (s *server) cleanupPreviousSession(previousSession *syncSession) {
	if !previousSession.completed {
		if s.metrics != nil {
			op, grp, sn, sr, st := s.resolveSessionLabels(previousSession)
			s.metrics.totalErr.Inc(1, op, grp, sn, sr, st, "switch")
		}
		previousSession.completed = true
	}

	if previousSession.partCtx == nil {
		return
	}
	if previousSession.partCtx.Handler != nil {
		if finishErr := previousSession.partCtx.Handler.FinishSync(); finishErr != nil {
			if s.metrics != nil {
				op, grp, sn, sr, st := s.resolveSessionLabels(previousSession)
				s.metrics.totalErr.Inc(1, op, grp, sn, sr, st, "finish_sync_err")
			}
			s.log.Error().Err(finishErr).Str("session_id", previousSession.sessionID).Msg("failed to finish sync for previous session")
		}
	}
	if closeErr := previousSession.partCtx.Close(); closeErr != nil {
		s.log.Error().Err(closeErr).Str("session_id", previousSession.sessionID).Msg("failed to close previous session partCtx")
	}
}

func (s *server) processChunk(stream clusterv1.ChunkedSyncService_SyncPartServer, session *syncSession, req *clusterv1.SyncPartRequest) error {
	// Check version compatibility on every chunk
	if req.VersionInfo != nil {
		versionCompatibility, syncStatus := checkSyncVersionCompatibility(req.VersionInfo)
		if syncStatus != clusterv1.SyncStatus_SYNC_STATUS_CHUNK_RECEIVED {
			s.log.Warn().
				Str("session_id", req.SessionId).
				Str("client_api_version", req.VersionInfo.ApiVersion).
				Str("client_file_format_version", req.VersionInfo.FileFormatVersion).
				Str("reason", versionCompatibility.Reason).
				Msg("sync version compatibility check failed")

			return s.sendResponse(stream, req, syncStatus, versionCompatibility.Reason, versionCompatibility)
		}
	}

	if !s.enableChunkReordering {
		return s.processChunkSequential(stream, session, req)
	}

	if session.chunkBuffer == nil {
		session.chunkBuffer = &chunkBuffer{
			chunks:        make(map[uint32]*clusterv1.SyncPartRequest),
			expectedIndex: 0,
			maxBufferSize: s.maxChunkBufferSize,
			bufferTimeout: s.chunkBufferTimeout,
			lastActivity:  time.Now(),
		}
	}

	return s.processChunkWithReordering(stream, session, req)
}

func (s *server) processChunkSequential(stream clusterv1.ChunkedSyncService_SyncPartServer, session *syncSession, req *clusterv1.SyncPartRequest) error {
	if req.ChunkIndex != session.chunksReceived {
		errMsg := fmt.Sprintf("out of order chunk received: expected %d, got %d",
			session.chunksReceived, req.ChunkIndex)
		s.log.Warn().Str("session_id", req.SessionId).
			Uint32("expected_chunk", session.chunksReceived).
			Uint32("received_chunk", req.ChunkIndex).
			Msg("out of order chunk received")
		return s.sendResponse(stream, req, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_OUT_OF_ORDER, errMsg, nil)
	}

	return s.processExpectedChunk(stream, session, req)
}

func (s *server) processChunkWithReordering(stream clusterv1.ChunkedSyncService_SyncPartServer, session *syncSession, req *clusterv1.SyncPartRequest) error {
	buffer := session.chunkBuffer
	// must check buffer timeout before refreshing lastActivity, otherwise it will never timeout.
	if checkErr := s.checkBufferTimeout(session); checkErr != nil {
		return checkErr
	}
	buffer.lastActivity = time.Now()

	if req.ChunkIndex == buffer.expectedIndex {
		if processErr := s.processExpectedChunk(stream, session, req); processErr != nil {
			return processErr
		}
		buffer.expectedIndex++

		return s.processBufferedChunks(stream, session)
	}

	if req.ChunkIndex > buffer.expectedIndex {
		gap := req.ChunkIndex - buffer.expectedIndex

		if gap > s.maxChunkGapSize {
			errMsg := fmt.Sprintf("chunk gap too large: expected %d, got %d (gap: %d > max: %d)",
				buffer.expectedIndex, req.ChunkIndex, gap, s.maxChunkGapSize)
			s.log.Warn().Str("session_id", req.SessionId).
				Uint32("expected_chunk", buffer.expectedIndex).
				Uint32("received_chunk", req.ChunkIndex).
				Uint32("gap", gap).
				Uint32("max_gap", s.maxChunkGapSize).
				Msg("chunk gap too large, rejecting")
			if s.metrics != nil {
				op, grp, sn, sr, st := s.resolveSessionLabels(session)
				s.metrics.totalErr.Inc(1, op, grp, sn, sr, st, "gap_too_large")
			}
			return s.sendResponse(stream, req, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_OUT_OF_ORDER, errMsg, nil)
		}

		if uint32(len(buffer.chunks)) >= buffer.maxBufferSize {
			errMsg := fmt.Sprintf("chunk buffer full, cannot store out-of-order chunk %d", req.ChunkIndex)
			s.log.Warn().Str("session_id", req.SessionId).
				Uint32("chunk_index", req.ChunkIndex).
				Uint32("buffer_size", uint32(len(buffer.chunks))).
				Uint32("max_buffer_size", buffer.maxBufferSize).
				Msg("chunk buffer full, rejecting chunk")
			if s.metrics != nil {
				op, grp, sn, sr, st := s.resolveSessionLabels(session)
				s.metrics.totalErr.Inc(1, op, grp, sn, sr, st, "buffer_full")
			}
			return s.sendResponse(stream, req, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_OUT_OF_ORDER, errMsg, nil)
		}

		buffer.chunks[req.ChunkIndex] = req
		if dl := s.log.Debug(); dl.Enabled() {
			dl.Str("session_id", req.SessionId).
				Str("topic", session.metadata.Topic).
				Uint32("chunk_index", req.ChunkIndex).
				Uint32("expected_index", buffer.expectedIndex).
				Uint32("buffered_chunks", uint32(len(buffer.chunks))).
				Msg("buffered out-of-order chunk")
		}
		return s.sendResponse(stream, req, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_RECEIVED,
			fmt.Sprintf("chunk %d buffered (waiting for %d)", req.ChunkIndex, buffer.expectedIndex), nil)
	}

	if req.ChunkIndex < buffer.expectedIndex {
		s.log.Warn().
			Str("session_id", req.SessionId).
			Str("topic", session.metadata.Topic).
			Uint32("chunk_index", req.ChunkIndex).
			Uint32("expected_index", buffer.expectedIndex).
			Msg("received duplicate or old chunk, ignoring")

		return s.sendResponse(stream, req, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_RECEIVED,
			fmt.Sprintf("duplicate chunk %d ignored", req.ChunkIndex), nil)
	}

	return nil
}

func (s *server) processExpectedChunk(stream clusterv1.ChunkedSyncService_SyncPartServer, session *syncSession, req *clusterv1.SyncPartRequest) error {
	calculatedChecksum := fmt.Sprintf("%x", crc32.ChecksumIEEE(req.ChunkData))
	if calculatedChecksum != req.ChunkChecksum {
		errMsg := fmt.Sprintf("chunk %d checksum mismatch: expected %s, got %s",
			req.ChunkIndex, req.ChunkChecksum, calculatedChecksum)
		s.log.Warn().Str("session_id", req.SessionId).Msg(errMsg)
		if s.metrics != nil {
			op, grp, sn, sr, st := s.resolveSessionLabels(session)
			s.metrics.totalErr.Inc(1, op, grp, sn, sr, st, "checksum_mismatch")
		}
		return s.sendResponse(stream, req, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_CHECKSUM_MISMATCH, errMsg, nil)
	}

	session.totalReceived += uint64(len(req.ChunkData))
	session.chunksReceived++

	var topic bus.Topic
	t, ok := data.TopicMap[session.metadata.Topic]
	if !ok {
		return fmt.Errorf("unknown sync topic: %s", session.metadata.Topic)
	}
	topic = t

	handler, exists := s.chunkedSyncHandlers[topic]
	if !exists {
		return fmt.Errorf("no handler registered for topic %s", topic)
	}

	for partIndex, partInfo := range req.PartsInfo {
		createNewContext := session.partCtx == nil ||
			session.partCtx.ID != partInfo.Id

		if createNewContext && session.partCtx != nil && session.partCtx.Handler != nil {
			if finishErr := session.partCtx.Handler.FinishSync(); finishErr != nil {
				return fmt.Errorf("failed to complete part %d: %w", session.partCtx.ID, finishErr)
			}
		}

		if createNewContext {
			session.partCtx = &queue.ChunkedSyncPartContext{
				Context:               stream.Context(),
				ID:                    partInfo.Id,
				Group:                 session.metadata.Group,
				ShardID:               session.metadata.ShardId,
				CompressedSizeBytes:   partInfo.CompressedSizeBytes,
				UncompressedSizeBytes: partInfo.UncompressedSizeBytes,
				TotalCount:            partInfo.TotalCount,
				BlocksCount:           partInfo.BlocksCount,
				MinTimestamp:          partInfo.MinTimestamp,
				MaxTimestamp:          partInfo.MaxTimestamp,
				MinKey:                partInfo.MinKey,
				MaxKey:                partInfo.MaxKey,
				PartType:              partInfo.PartType,
			}
			partHandler, createErr := handler.CreatePartHandler(session.partCtx)
			if createErr != nil {
				return fmt.Errorf("failed to create part handler: %w", createErr)
			}
			session.partCtx.Handler = partHandler
		} else if session.partCtx.PartType != partInfo.PartType {
			session.partCtx.CompressedSizeBytes = partInfo.CompressedSizeBytes
			session.partCtx.UncompressedSizeBytes = partInfo.UncompressedSizeBytes
			session.partCtx.TotalCount = partInfo.TotalCount
			session.partCtx.BlocksCount = partInfo.BlocksCount
			session.partCtx.MinTimestamp = partInfo.MinTimestamp
			session.partCtx.MaxTimestamp = partInfo.MaxTimestamp
			session.partCtx.MinKey = partInfo.MinKey
			session.partCtx.MaxKey = partInfo.MaxKey
			session.partCtx.PartType = partInfo.PartType
			if newPartErr := session.partCtx.Handler.NewPartType(session.partCtx); newPartErr != nil {
				return fmt.Errorf("failed to new part type: %w", newPartErr)
			}
		}

		if processErr := s.processPart(session, req, partInfo, partIndex, handler); processErr != nil {
			if errors.Is(processErr, queue.ErrServerBusy) {
				return s.sendResponse(stream, req, clusterv1.SyncStatus_SYNC_STATUS_SERVER_BUSY,
					"receiver under memory pressure, retry later", nil)
			}
			s.log.Error().Err(processErr).
				Str("session_id", req.SessionId).
				Str("topic", session.metadata.Topic).
				Int("part_index", partIndex).
				Msg("failed to process part")
			if s.metrics != nil {
				op, grp, sn, sr, st := s.resolveSessionLabels(session)
				s.metrics.totalErr.Inc(1, op, grp, sn, sr, st, "part_failed")
			}
			return processErr
		}
	}

	return s.sendResponse(stream, req, clusterv1.SyncStatus_SYNC_STATUS_CHUNK_RECEIVED, "", nil)
}

func (s *server) processBufferedChunks(stream clusterv1.ChunkedSyncService_SyncPartServer, session *syncSession) error {
	buffer := session.chunkBuffer

	for {
		if chunk, exists := buffer.chunks[buffer.expectedIndex]; exists {
			delete(buffer.chunks, buffer.expectedIndex)

			if dl := s.log.Debug(); dl.Enabled() {
				dl.Str("session_id", session.sessionID).
					Str("topic", session.metadata.Topic).
					Uint32("chunk_index", buffer.expectedIndex).
					Uint32("remaining_buffered", uint32(len(buffer.chunks))).
					Msg("processing buffered chunk")
			}

			if processErr := s.processExpectedChunk(stream, session, chunk); processErr != nil {
				return processErr
			}
			buffer.expectedIndex++
		} else {
			break
		}
	}

	return nil
}

func (s *server) checkBufferTimeout(session *syncSession) error {
	if session.chunkBuffer == nil {
		return nil
	}

	if time.Since(session.chunkBuffer.lastActivity) > session.chunkBuffer.bufferTimeout {
		if len(session.chunkBuffer.chunks) > 0 {
			missing := make([]uint32, 0)
			for i := session.chunkBuffer.expectedIndex; i < session.chunkBuffer.expectedIndex+10; i++ {
				if _, exists := session.chunkBuffer.chunks[i]; !exists {
					missing = append(missing, i)
				}
			}
			if s.metrics != nil {
				op, grp, sn, sr, st := s.resolveSessionLabels(session)
				s.metrics.totalErr.Inc(1, op, grp, sn, sr, st, "buffer_timeout")
			}
			return fmt.Errorf("buffer timeout: missing chunks %v after %v",
				missing, session.chunkBuffer.bufferTimeout)
		}
	}

	return nil
}

func (s *server) processPart(session *syncSession, req *clusterv1.SyncPartRequest, partInfo *clusterv1.PartInfo, partIndex int, handler queue.ChunkedSyncHandler) error {
	var totalPartSizeInChunk uint32
	for _, fileInfo := range partInfo.Files {
		totalPartSizeInChunk += fileInfo.Size
	}
	progress := session.partsProgress[partIndex]
	if progress == nil {
		session.partsProgress[partIndex] = &partProgress{
			totalBytes:    totalPartSizeInChunk,
			receivedBytes: 0,
			completed:     false,
		}
		progress = session.partsProgress[partIndex]
	} else {
		progress.completed = false
		progress.totalBytes += totalPartSizeInChunk
	}

	var partDataSize uint32
	for _, fileInfo := range partInfo.Files {
		if fileInfo.Offset >= uint32(len(req.ChunkData)) {
			s.log.Warn().Str("session_id", session.sessionID).
				Str("topic", session.metadata.Topic).
				Str("file_name", fileInfo.Name).
				Uint32("offset", fileInfo.Offset).
				Uint32("chunk_size", uint32(len(req.ChunkData))).
				Msg("file data not in this chunk")
			continue
		}

		fileEndOffset := fileInfo.Offset + fileInfo.Size
		actualFileSize := fileInfo.Size
		if fileEndOffset > uint32(len(req.ChunkData)) {
			actualFileSize = uint32(len(req.ChunkData)) - fileInfo.Offset
		}

		fileChunk := req.ChunkData[fileInfo.Offset : fileInfo.Offset+actualFileSize]
		partDataSize += actualFileSize

		session.partCtx.FileName = fileInfo.Name
		session.partCtx.PartType = partInfo.PartType

		if handleErr := handler.HandleFileChunk(session.partCtx, fileChunk); handleErr != nil {
			return fmt.Errorf("failed to stream file chunk for %s: %w", fileInfo.Name, handleErr)
		}
	}

	progress.receivedBytes += partDataSize
	if progress.receivedBytes > progress.totalBytes {
		return fmt.Errorf("received more bytes (%d) than expected (%d) for part %d",
			progress.receivedBytes, progress.totalBytes, partIndex)
	}

	progress.completed = progress.receivedBytes == progress.totalBytes
	return nil
}

func (s *server) handleCompletion(stream clusterv1.ChunkedSyncService_SyncPartServer, session *syncSession, req *clusterv1.SyncPartRequest) error {
	if session.partCtx != nil && session.partCtx.Handler != nil {
		if finishErr := session.partCtx.Handler.FinishSync(); finishErr != nil {
			if s.metrics != nil {
				op, grp, sn, sr, st := s.resolveSessionLabels(session)
				s.metrics.totalErr.Inc(1, op, grp, sn, sr, st, "finish_sync_err")
			}
			return fmt.Errorf("failed to complete part %d: %w", session.partCtx.ID, finishErr)
		}
		session.partCtx.Handler = nil
	}

	partsResults := make([]*clusterv1.PartResult, 0, len(session.partsProgress))
	allPartsSuccessful := true

	for _, progress := range session.partsProgress {
		success := progress.completed
		if !success {
			allPartsSuccessful = false
		}

		partsResults = append(partsResults, &clusterv1.PartResult{
			Success:        success,
			Error:          session.errorMsg,
			BytesProcessed: progress.receivedBytes,
		})
	}

	syncResult := &clusterv1.SyncResult{
		Success:            allPartsSuccessful,
		TotalBytesReceived: session.totalReceived,
		DurationMs:         time.Since(session.startTime).Milliseconds(),
		ChunksReceived:     session.chunksReceived,
		PartsReceived:      uint32(len(session.partsProgress)),
		PartsResults:       partsResults,
	}

	session.completed = true
	if s.metrics != nil {
		op, grp, sn, sr, st := s.resolveSessionLabels(session)
		s.metrics.totalFinished.Inc(1, op, grp, sn, sr, st)
		s.metrics.totalLatency.Observe(float64(syncResult.DurationMs)/1000.0, op, grp, sn, sr, st)
		s.metrics.receivedBytes.Inc(float64(syncResult.TotalBytesReceived), op, grp, sn, sr, st)
		for _, pr := range partsResults {
			if !pr.Success {
				s.metrics.totalErr.Inc(1, op, grp, sn, sr, st, "part_failed")
			}
		}
	}

	if dl := s.log.Debug(); dl.Enabled() {
		dl.Str("session_id", session.sessionID).
			Str("topic", session.metadata.Topic).
			Bool("success", syncResult.Success).
			Uint64("bytes_received", syncResult.TotalBytesReceived).
			Int64("duration_ms", syncResult.DurationMs).
			Msg("completed chunked sync session")
	}

	return s.sendResponse(stream, req, clusterv1.SyncStatus_SYNC_STATUS_SYNC_COMPLETE, "", syncResult)
}

func (s *server) sendResponse(
	stream clusterv1.ChunkedSyncService_SyncPartServer,
	req *clusterv1.SyncPartRequest,
	syncStatus clusterv1.SyncStatus,
	errorMsg string,
	result interface{},
) error {
	resp := &clusterv1.SyncPartResponse{
		SessionId:  req.SessionId,
		ChunkIndex: req.ChunkIndex,
		Status:     syncStatus,
		Error:      errorMsg,
	}

	switch r := result.(type) {
	case *clusterv1.SyncResult:
		resp.SyncResult = r
	case *clusterv1.VersionCompatibility:
		resp.VersionCompatibility = r
	}

	return stream.Send(resp)
}
