Move service related filed to services folder

services/credentials/repositories/grpc_client.go

@@ -0,0 +1,136 @@
+package repositories
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"fmt"
+	"io"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+
+	"github.com/mitchell/selfpass/services/credentials/endpoints"
+	"github.com/mitchell/selfpass/services/credentials/protobuf"
+	"github.com/mitchell/selfpass/services/credentials/transport"
+	"github.com/mitchell/selfpass/services/credentials/types"
+)
+
+func NewCredentialServiceClient(ctx context.Context, target, ca, cert, key string) (types.CredentialClient, error) {
+	keypair, err := tls.X509KeyPair([]byte(cert), []byte(key))
+	if err != nil {
+		return nil, err
+	}
+
+	capool := x509.NewCertPool()
+	capool.AppendCertsFromPEM([]byte(ca))
+
+	creds := credentials.NewTLS(&tls.Config{
+		RootCAs:      capool,
+		Certificates: []tls.Certificate{keypair},
+		MinVersion:   tls.VersionTLS12,
+		CurvePreferences: []tls.CurveID{
+			tls.CurveP256,
+		},
+	})
+
+	conn, err := grpc.DialContext(ctx, target, grpc.WithTransportCredentials(creds), grpc.WithBlock())
+	if err != nil {
+		return nil, err
+	}
+
+	return credentialServiceClient{
+		client: protobuf.NewCredentialServiceClient(conn),
+	}, nil
+}
+
+type credentialServiceClient struct {
+	client protobuf.CredentialServiceClient
+}
+
+func (c credentialServiceClient) GetAllMetadata(ctx context.Context, sourceHost string) (output <-chan types.Metadata, errch chan error) {
+	pbmdch := make(chan protobuf.Metadata, 1)
+	errch = make(chan error, 1)
+
+	stream, err := transport.DecodeMetdataStreamResponse(ctx, transport.ProtobufMetadataStream{
+		Metadata: pbmdch,
+		Errors:   errch,
+	})
+
+	srv, err := c.client.GetAllMetadata(ctx, &protobuf.GetAllMetadataRequest{SourceHost: sourceHost})
+	if err != nil {
+		errch <- err
+		return nil, errch
+	}
+
+	go func() {
+		defer close(pbmdch)
+
+		for {
+			select {
+			case <-ctx.Done():
+				errch <- fmt.Errorf("context timeout")
+				return
+			default:
+			}
+
+			pbmd, err := srv.Recv()
+			if err == io.EOF {
+				return
+			} else if err != nil {
+				errch <- err
+				return
+			}
+
+			pbmdch <- *pbmd
+		}
+	}()
+
+	return stream.Metadata, stream.Errors
+}
+
+func (c credentialServiceClient) Get(ctx context.Context, id string) (output types.Credential, err error) {
+	req := transport.EncodeIdRequest(endpoints.IDRequest{ID: id})
+
+	res, err := c.client.Get(ctx, &req)
+	if err != nil {
+		return output, err
+	}
+
+	return transport.DecodeCredential(*res)
+}
+
+func (c credentialServiceClient) Create(ctx context.Context, ci types.CredentialInput) (output types.Credential, err error) {
+	req := transport.EncodeCredentialRequest(ci)
+
+	res, err := c.client.Create(ctx, &req)
+	if err != nil {
+		return output, err
+	}
+
+	return transport.DecodeCredential(*res)
+}
+
+func (c credentialServiceClient) Update(ctx context.Context, id string, ci types.CredentialInput) (output types.Credential, err error) {
+	req := transport.EncodeUpdateRequest(endpoints.UpdateRequest{ID: id, Credential: ci})
+
+	res, err := c.client.Update(ctx, &req)
+	if err != nil {
+		return output, err
+	}
+
+	return transport.DecodeCredential(*res)
+}
+
+func (c credentialServiceClient) Delete(ctx context.Context, id string) (err error) {
+	req := transport.EncodeIdRequest(endpoints.IDRequest{ID: id})
+
+	res, err := c.client.Delete(ctx, &req)
+	if err != nil {
+		return err
+	}
+	if !res.Success {
+		return fmt.Errorf("delete unsuccessful")
+	}
+	return nil
+}

services/credentials/repositories/redis.go

@@ -0,0 +1,81 @@
+package repositories
+
+import (
+	"context"
+
+	"github.com/mediocregopher/radix/v3"
+	"github.com/mitchell/selfpass/services/credentials/types"
+)
+
+func NewRedisConn(networkType, address string, connCount uint, options ...radix.PoolOpt) (c RedisConn, err error) {
+	p, err := radix.NewPool(networkType, address, int(connCount), options...)
+	return RedisConn{p: p}, err
+}
+
+type RedisConn struct {
+	p *radix.Pool
+}
+
+func (conn RedisConn) GetAllMetadata(ctx context.Context, sourceHost string, errch chan<- error) (output <-chan types.Metadata) {
+	mdch := make(chan types.Metadata, 1)
+
+	go func() {
+		defer close(mdch)
+
+		var key string
+		scr := radix.NewScanner(conn.p, radix.ScanOpts{Command: scan, Pattern: types.TypePrefixCred + dash + sourceHost + star})
+
+		for scr.Next(&key) {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+			}
+
+			var md types.Metadata
+
+			if err := conn.p.Do(radix.Cmd(&md, hGetAll, key)); err != nil {
+				errch <- err
+				return
+			}
+
+			mdch <- md
+		}
+	}()
+
+	return mdch
+}
+
+func (conn RedisConn) Get(ctx context.Context, id string) (output types.Credential, err error) {
+	err = conn.p.Do(radix.Cmd(&output, hGetAll, id))
+	return output, err
+}
+
+func (conn RedisConn) Put(ctx context.Context, c types.Credential) (err error) {
+	err = conn.p.Do(radix.FlatCmd(nil, hMSet, c.ID, c))
+	return err
+}
+
+func (conn RedisConn) Delete(ctx context.Context, id string) (err error) {
+	err = conn.p.Do(radix.Cmd(nil, del, id))
+	return err
+}
+
+func (conn RedisConn) DumpDB(ctx context.Context) (bs []byte, err error) {
+	bs = []byte{}
+
+	if err := conn.p.Do(radix.Cmd(&bs, "DUMP")); err != nil {
+		return nil, err
+	}
+
+	return bs, nil
+}
+
+const (
+	dash    = "-"
+	star    = "*"
+	scan    = "SCAN"
+	hGetAll = "HGETALL"
+	hMSet   = "HMSET"
+	del     = "DEL"
+)