package storage

import (
	"errors"

	riak "github.com/basho/riak-go-client"
)

type Riak struct {
	db *riak.Client
}

func NewRiak(addr string, addrs ...string) (*Riak, error) {
	clientOpts := &riak.NewClientOptions{
		RemoteAddresses: append(addrs, addr),
	}

	db, err := riak.NewClient(clientOpts)
	if err != nil {
		return nil, err
	}

	ok, err := db.Ping()
	if !ok {
		return nil, ErrCantConnect
	}

	return &Riak{
		db: db,
	}, err
}

func (r *Riak) List(ns []string, limits ...string) ([]string, error) {
	return nil, errors.New("not impl")
}

func (r *Riak) Get(key string, ns ...string) ([]byte, error) {
	namespace := resolveNamespace(ns)
	obj := &riak.Object{}

	cmd, err := riak.NewFetchValueCommandBuilder().
		WithBucket(namespace).
		WithKey(key).
		Build()
	if err != nil {
		return nil, err
	}

	err = r.db.Execute(cmd)
	return obj.Value, err
}

func (r *Riak) Set(key string, value []byte, ns ...string) error {
	namespace := resolveNamespace(ns)
	obj := &riak.Object{
		Value: value,
	}

	cmd, err := riak.NewStoreValueCommandBuilder().
		WithBucket(namespace).
		WithContent(obj).
		Build()
	if err != nil {
		return err
	}

	return r.db.Execute(cmd)
}

func (r *Riak) Close() error {
	return r.db.Stop()
}