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) 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} } } }