Agate/remote/client.go

package remote

import (
	"context"
	"fmt"
	"io"
	"os"
	"path/filepath"

	stdgrpc "google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"

	agateGrpc "gitea.unprism.ru/KRBL/Agate/grpc"
	"gitea.unprism.ru/KRBL/Agate/hash"
	"gitea.unprism.ru/KRBL/Agate/interfaces"
	"gitea.unprism.ru/KRBL/Agate/store"
)

// Client представляет клиент для подключения к удаленному серверу снапшотов.
type Client struct {
	conn   *stdgrpc.ClientConn
	client agateGrpc.SnapshotServiceClient
}

// Убедимся, что Client реализует интерфейс interfaces.SnapshotClient
var _ interfaces.SnapshotClient = (*Client)(nil)

// NewClient создает нового клиента, подключенного к указанному адресу.
func NewClient(address string) (*Client, error) {
	conn, err := stdgrpc.Dial(address, stdgrpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		return nil, fmt.Errorf("failed to connect to server at %s: %w", address, err)
	}
	client := agateGrpc.NewSnapshotServiceClient(conn)
	return &Client{conn: conn, client: client}, nil
}

// Close закрывает соединение с сервером.
func (c *Client) Close() error {
	if c.conn != nil {
		return c.conn.Close()
	}
	return nil
}

// ListSnapshots получает список снапшотов с удаленного сервера.
func (c *Client) ListSnapshots(ctx context.Context) ([]store.SnapshotInfo, error) {
	response, err := c.client.ListSnapshots(ctx, &agateGrpc.ListSnapshotsRequest{})
	if err != nil {
		return nil, fmt.Errorf("failed to list snapshots: %w", err)
	}

	snapshots := make([]store.SnapshotInfo, 0, len(response.Snapshots))
	for _, snapshot := range response.Snapshots {
		snapshots = append(snapshots, store.SnapshotInfo{
			ID:           snapshot.Id,
			Name:         snapshot.Name,
			ParentID:     snapshot.ParentId,
			CreationTime: snapshot.CreationTime.AsTime(),
		})
	}
	return snapshots, nil
}

// FetchSnapshotDetails получает детальную информацию о конкретном снапшоте.
func (c *Client) FetchSnapshotDetails(ctx context.Context, snapshotID string) (*store.Snapshot, error) {
	response, err := c.client.GetSnapshotDetails(ctx, &agateGrpc.GetSnapshotDetailsRequest{
		SnapshotId: snapshotID,
	})
	if err != nil {
		return nil, fmt.Errorf("failed to get snapshot details: %w", err)
	}

	snapshot := &store.Snapshot{
		ID:           response.Info.Id,
		Name:         response.Info.Name,
		ParentID:     response.Info.ParentId,
		CreationTime: response.Info.CreationTime.AsTime(),
		Files:        make([]store.FileInfo, 0, len(response.Files)),
	}

	for _, file := range response.Files {
		snapshot.Files = append(snapshot.Files, store.FileInfo{
			Path:   file.Path,
			Size:   file.SizeBytes,
			IsDir:  file.IsDir,
			SHA256: file.Sha256Hash,
		})
	}
	return snapshot, nil
}

// GetDiffInfo gets the hash and size of a differential archive.
func (c *Client) GetDiffInfo(ctx context.Context, snapshotID, localParentID string) (*store.DiffInfo, error) {
	req := &agateGrpc.GetDiffInfoRequest{
		SnapshotId:    snapshotID,
		LocalParentId: localParentID,
	}

	info, err := c.client.GetDiffInfo(ctx, req)
	if err != nil {
		return nil, fmt.Errorf("failed to get diff info: %w", err)
	}

	return &store.DiffInfo{
		SHA256: info.Sha256Hash,
		Size:   info.SizeBytes,
	}, nil
}

// DownloadSnapshotDiff скачивает архив с разницей между снапшотами.
func (c *Client) DownloadSnapshotDiff(ctx context.Context, snapshotID, localParentID, targetPath string) error {
	// Check for local file and validate it
	if fileInfo, err := os.Stat(targetPath); err == nil {
		remoteDiffInfo, err := c.GetDiffInfo(ctx, snapshotID, localParentID)
		if err != nil {
			// Log the error but proceed with download
			fmt.Printf("could not get remote diff info: %v. proceeding with download.", err)
		} else {
			if fileInfo.Size() == remoteDiffInfo.Size {
				localHash, err := hash.CalculateFileHash(targetPath)
				if err == nil && localHash == remoteDiffInfo.SHA256 {
					fmt.Printf("local snapshot archive %s is valid, skipping download.", targetPath)
					return nil // File is valid, skip download
				}
			}
		}
	}

	var offset int64
	fileInfo, err := os.Stat(targetPath)
	if err == nil {
		offset = fileInfo.Size()
	} else if !os.IsNotExist(err) {
		return fmt.Errorf("failed to stat temporary file: %w", err)
	}

	req := &agateGrpc.DownloadSnapshotDiffRequest{
		SnapshotId:    snapshotID,
		LocalParentId: localParentID,
		Offset:        offset,
	}

	stream, err := c.client.DownloadSnapshotDiff(ctx, req)
	if err != nil {
		return fmt.Errorf("failed to start snapshot diff download: %w", err)
	}

	if err := os.MkdirAll(filepath.Dir(targetPath), 0755); err != nil {
		return fmt.Errorf("failed to create directory for %s: %w", targetPath, err)
	}

	file, err := os.OpenFile(targetPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644)
	if err != nil {
		return fmt.Errorf("failed to open file %s: %w", targetPath, err)
	}
	defer func() {
		if err := file.Close(); err != nil {
			fmt.Printf("failed to close file: %v", err)
		}
	}()

	for {
		resp, err := stream.Recv()
		if err == io.EOF {
			break
		}
		if err != nil {
			return fmt.Errorf("error receiving diff chunk: %w", err)
		}
		if _, err := file.Write(resp.ChunkData); err != nil {
			return fmt.Errorf("error writing to file: %w", err)
		}
	}

	return nil
}