zw-go/client.go

package zw

import (
	"context"
	"errors"
	"fmt"
	"io"
	"net/url"

	zwdaemonv1 "git.pablu.de/pablu/zw-go/gen/zwdaemon/v1"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
)

const supportedAPIMajor uint32 = 1

func optionalString(value string) *string {
	if value == "" {
		return nil
	}
	return &value
}

type Client struct {
	conn *grpc.ClientConn
	rpc  zwdaemonv1.ZwDaemonServiceClient
}

func Connect(ctx context.Context, socketPath string) (*Client, error) {
	target := (&url.URL{
		Scheme: "unix",
		Path:   socketPath,
	}).String()

	conn, err := grpc.NewClient(
		target,
		grpc.WithTransportCredentials(insecure.NewCredentials()),
	)
	if err != nil {
		return nil, fmt.Errorf("create daemon client: %w", err)
	}

	client := &Client{
		conn: conn,
		rpc:  zwdaemonv1.NewZwDaemonServiceClient(conn),
	}

	info, err := client.ServerInfo(ctx)
	if err != nil {
		_ = conn.Close()
		return nil, err
	}

	if info.APIMajor != supportedAPIMajor {
		_ = conn.Close()
		return nil, &IncompatibleAPIVersionError{
			ServerMajor:    info.APIMajor,
			ServerMinor:    info.APIMinor,
			SupportedMajor: supportedAPIMajor,
		}
	}

	return client, nil
}

func (c *Client) Close() error {
	return c.conn.Close()
}

func (c *Client) ServerInfo(ctx context.Context) (ServerInfo, error) {
	response, err := c.rpc.GetServerInfo(ctx, &zwdaemonv1.GetServerInfoRequest{})
	if err != nil {
		return ServerInfo{}, convertRPCError(err)
	}

	return ServerInfo{
		APIMajor:      response.ApiMajor,
		APIMinor:      response.ApiMinor,
		DaemonVersion: response.DaemonVersion,
	}, nil
}

func (c *Client) Create(ctx context.Context, vaultPath string, password string) error {
	_, err := c.rpc.Create(ctx, &zwdaemonv1.CreateRequest{
		VaultPath: vaultPath,
		Password:  password,
	})
	if err != nil {
		return convertRPCError(err)
	}
	return nil
}

func (c *Client) Unseal(ctx context.Context, vaultPath string, password string) error {
	_, err := c.rpc.Unseal(ctx, &zwdaemonv1.UnsealRequest{
		VaultPath: vaultPath,
		Password:  password,
	})
	if err != nil {
		return convertRPCError(err)
	}
	return nil
}

func (c *Client) Seal(ctx context.Context) error {
	_, err := c.rpc.Seal(ctx, &zwdaemonv1.SealRequest{})
	if err != nil {
		return convertRPCError(err)
	}
	return nil
}

func (c *Client) ListFiles(ctx context.Context, filter ListFilter) ([]ListedFile, error) {
	response, err := c.rpc.ListFiles(ctx, &zwdaemonv1.ListFilesRequest{
		Prefix:   optionalString(filter.Prefix),
		Contains: optionalString(filter.Contains),
	})
	if err != nil {
		return nil, convertRPCError(err)
	}

	files := make([]ListedFile, len(response.GetFiles()))
	for index, file := range response.GetFiles() {
		files[index] = ListedFile{
			LogicalPath: file.GetLogicalPath(),
			Size:        file.GetSize(),
		}
	}

	return files, nil
}

func (c *Client) ListPaths(ctx context.Context) ([]string, error) {
	files, err := c.ListFiles(ctx, ListFilter{})
	if err != nil {
		return nil, err
	}

	paths := make([]string, len(files))
	for index, file := range files {
		paths[index] = file.LogicalPath
	}

	return paths, nil
}

func (c *Client) DeleteFile(ctx context.Context, logicalPath string) error {
	_, err := c.rpc.DeleteFile(ctx, &zwdaemonv1.DeleteFileRequest{
		LogicalPath: logicalPath,
	})
	if err != nil {
		return convertRPCError(err)
	}
	return nil
}

func (c *Client) MoveFile(
	ctx context.Context,
	sourcePath string,
	destinationPath string,
) error {
	_, err := c.rpc.MoveFile(ctx, &zwdaemonv1.MoveFileRequest{
		SourcePath:      sourcePath,
		DestinationPath: destinationPath,
	})
	if err != nil {
		return convertRPCError(err)
	}
	return nil
}

func (c *Client) CopyFile(
	ctx context.Context,
	sourcePath string,
	destinationPath string,
) error {
	_, err := c.rpc.CopyFile(ctx, &zwdaemonv1.CopyFileRequest{
		SourcePath:      sourcePath,
		DestinationPath: destinationPath,
	})
	if err != nil {
		return convertRPCError(err)
	}
	return nil
}

func (c *Client) Compact(ctx context.Context) error {
	_, err := c.rpc.Compact(ctx, &zwdaemonv1.CompactRequest{})
	if err != nil {
		return convertRPCError(err)
	}
	return nil
}

func (c *Client) GetFile(
	ctx context.Context,
	logicalPath string,
) (io.ReadCloser, error) {
	streamCtx, cancel := context.WithCancel(ctx)

	stream, err := c.rpc.GetFile(streamCtx, &zwdaemonv1.GetFileRequest{
		LogicalPath: logicalPath,
	})
	if err != nil {
		cancel()
		return nil, convertRPCError(err)
	}

	return &downloadReader{
		cancel: cancel,
		recv: func() ([]byte, error) {
			response, err := stream.Recv()
			if err != nil {
				return nil, err
			}
			return response.GetData(), nil
		},
	}, nil
}

func (c *Client) GetFileRange(
	ctx context.Context,
	logicalPath string,
	start uint64,
	exclusiveEnd uint64,
) (io.ReadCloser, error) {
	streamCtx, cancel := context.WithCancel(ctx)

	stream, err := c.rpc.GetFileRange(streamCtx, &zwdaemonv1.GetFileRangeRequest{
		LogicalPath:  logicalPath,
		Start:        start,
		ExclusiveEnd: exclusiveEnd,
	})
	if err != nil {
		cancel()
		return nil, convertRPCError(err)
	}

	return &downloadReader{
		cancel: cancel,
		recv: func() ([]byte, error) {
			response, err := stream.Recv()
			if err != nil {
				return nil, err
			}
			return response.GetData(), nil
		},
	}, nil
}

func (c *Client) AddFile(
	ctx context.Context,
	logicalPath string,
	source io.Reader,
	sourceLen uint64,
) error {
	uploadCtx, cancel := context.WithCancel(ctx)
	defer cancel()

	stream, err := c.rpc.AddFile(uploadCtx)
	if err != nil {
		return convertRPCError(err)
	}

	send := func(request *zwdaemonv1.AddFileRequest) error {
		err := stream.Send(request)
		if err == nil {
			return nil
		}

		if errors.Is(err, io.EOF) {
			_, receiveErr := stream.CloseAndRecv()
			if receiveErr != nil {
				return convertRPCError(receiveErr)
			}
		}

		return convertRPCError(err)
	}

	err = send(&zwdaemonv1.AddFileRequest{
		Payload: &zwdaemonv1.AddFileRequest_Header{
			Header: &zwdaemonv1.AddFileHeader{
				LogicalPath: logicalPath,
				SourceLen:   sourceLen,
			},
		},
	})
	if err != nil {
		return err
	}

	const chunkSize = 512 * 1024
	buffer := make([]byte, chunkSize)

	for {
		n, readErr := source.Read(buffer)

		if n > 0 {
			err := send(&zwdaemonv1.AddFileRequest{
				Payload: &zwdaemonv1.AddFileRequest_Chunk{
					Chunk: buffer[:n],
				},
			})
			if err != nil {
				return err
			}
		}

		switch {
		case readErr == nil:
			continue
		case errors.Is(readErr, io.EOF):
			_, err := stream.CloseAndRecv()
			if err != nil {
				return convertRPCError(err)
			}
			return nil
		default:
			return &UploadReadError{Err: readErr}
		}
	}
}

func (c *Client) ReplaceFile(
	ctx context.Context,
	logicalPath string,
	source io.Reader,
	sourceLen uint64,
) error {
	uploadCtx, cancel := context.WithCancel(ctx)
	defer cancel()

	stream, err := c.rpc.ReplaceFile(uploadCtx)
	if err != nil {
		return convertRPCError(err)
	}

	send := func(request *zwdaemonv1.ReplaceFileRequest) error {
		err := stream.Send(request)
		if err == nil {
			return nil
		}

		if errors.Is(err, io.EOF) {
			_, receiveErr := stream.CloseAndRecv()
			if receiveErr != nil {
				return convertRPCError(receiveErr)
			}
		}

		return convertRPCError(err)
	}

	err = send(&zwdaemonv1.ReplaceFileRequest{
		Payload: &zwdaemonv1.ReplaceFileRequest_Header{
			Header: &zwdaemonv1.ReplaceFileHeader{
				LogicalPath: logicalPath,
				SourceLen:   sourceLen,
			},
		},
	})
	if err != nil {
		return err
	}

	const chunkSize = 512 * 1024
	buffer := make([]byte, chunkSize)

	for {
		n, readErr := source.Read(buffer)

		if n > 0 {
			err := send(&zwdaemonv1.ReplaceFileRequest{
				Payload: &zwdaemonv1.ReplaceFileRequest_Chunk{
					Chunk: buffer[:n],
				},
			})
			if err != nil {
				return err
			}
		}

		switch {
		case readErr == nil:
			continue
		case errors.Is(readErr, io.EOF):
			_, err := stream.CloseAndRecv()
			if err != nil {
				return convertRPCError(err)
			}
			return nil
		default:
			return &UploadReadError{Err: readErr}
		}
	}
}