query-orchestration/internal/client/sync/sync.go

package clientsync

import (
	"context"
	"database/sql"
	"errors"
	"log/slog"

	docsyncrunner "queryorchestration/api/docSyncRunner"
	"queryorchestration/internal/database"
	"queryorchestration/internal/database/repository"
	"queryorchestration/internal/serviceconfig/queue"

	"github.com/google/uuid"
)

func (s *Service) Sync(ctx context.Context, id uuid.UUID) error {
	hasMore := true
	offset := int32(0)
	dbid := database.MustToDBUUID(id)

	for hasMore {
		ids, err := s.cfg.GetDBQueries().ListDocumentIDsBatch(ctx, &repository.ListDocumentIDsBatchParams{
			Batchsize:  s.batchSize,
			Pageoffset: offset,
			Clientid:   dbid,
		})
		if errors.Is(err, sql.ErrNoRows) {
			return nil
		} else if err != nil {
			return err
		}

		slog.Debug("batch processing", "id", id, "offset", offset, "length", len(ids))

		for _, id := range ids {
			err := s.cfg.SendToQueue(ctx, &queue.SendParams{
				QueueURL: s.cfg.GetDocumentSyncURL(),
				Body: docsyncrunner.Body{
					ID: database.MustToUUID(id.ID),
				},
			})
			if err != nil {
				return err
			}
		}

		offset += s.batchSize
		if len(ids) == 0 || ids[0].Totalcount == nil || int64(offset) >= *ids[0].Totalcount {
			hasMore = false
		}
	}

	return nil
}