blob.table.go

package dbkit_tests

import (
	"bytes"
	"context"
	"strconv"

	"github.com/gocql/gocql"
	"github.com/oliverkofoed/gokit/logkit"
)

// BlobsTable is generated by dbkit
type BlobsTable struct {
	driver blobsDriver
}

// Insert creates a record in the Blobs table
func (t BlobsTable) Insert(ctx context.Context, id []byte, _type int32, data []byte) (*Blob, error) {
	return t.driver.insert(ctx, id, _type, data)
}

// Load a single record from the Blobs table based on the given query
func (t BlobsTable) Load(ctx context.Context, query string, args ...interface{}) (*Blob, error) {
	return t.driver.load(ctx, query, args...)
}

// Delete deletes records from the Blobs table based on the given query
func (t BlobsTable) Delete(ctx context.Context, query string, args ...interface{}) error {
	return t.driver.delete(ctx, query, args...)
}

// LoadByID loads a single record from the Blobs table based on the given values
func (t BlobsTable) LoadByID(ctx context.Context, id []byte) (*Blob, error) {
	return t.driver.loadByID(ctx, id)
}

// FindByID finds records the Blobs table based on the given values
func (t BlobsTable) FindByID(id []byte) *BlobQuery {
	return t.driver.findByID(id)
}

// DeleteByID deletes records the Blobs table based on the given values
func (t BlobsTable) DeleteByID(ctx context.Context, id []byte) error {
	return t.driver.deleteByID(ctx, id)
}

// LoadByIDAndType loads a single record from the Blobs table based on the given values
func (t BlobsTable) LoadByIDAndType(ctx context.Context, id []byte, _type int32) (*Blob, error) {
	return t.driver.loadByIDAndType(ctx, id, _type)
}

// FindByIDAndType finds records the Blobs table based on the given values
func (t BlobsTable) FindByIDAndType(id []byte, _type int32) *BlobQuery {
	return t.driver.findByIDAndType(id, _type)
}

// DeleteByIDAndType deletes records the Blobs table based on the given values
func (t BlobsTable) DeleteByIDAndType(ctx context.Context, id []byte, _type int32) error {
	return t.driver.deleteByIDAndType(ctx, id, _type)
}

// Query creates a query for records in the Blobs table by id
func (t BlobsTable) Query() *BlobQuery {
	return &BlobQuery{driver: t.driver}
}

// Blob represents a row in the Blobs table
type Blob struct {
	driver blobsDriver
	ID     []byte
	Type   int32
	Data   []byte

	loadID   []byte
	loadType int32
	loadData []byte
}

// Save saves any changes to the row
func (i *Blob) Save(ctx context.Context) error {
	return i.driver.save(ctx, i)
}

// BlobQuery is a modifiable query against the Blobs table
type BlobQuery struct {
	driver blobsDriver

	whereQuery     string
	orderColumns   []string
	orderDirection []int
	args           []interface{}
	limit          int
}

// Limit changes the limit of the BlobQuery query to the given value
func (q *BlobQuery) Limit(limit int) *BlobQuery {
	q.limit = limit
	return q
}

// SortAscendingBy adds an ascending sort column to the query
func (q *BlobQuery) SortAscendingBy(column string) *BlobQuery {
	q.orderColumns = append(q.orderColumns, column)
	q.orderDirection = append(q.orderDirection, 1)
	return q
}

// SortDescendingBy adds an descending sort column to the query
func (q *BlobQuery) SortDescendingBy(column string) *BlobQuery {
	q.orderColumns = append(q.orderColumns, column)
	q.orderDirection = append(q.orderDirection, -1)
	return q
}

// Where changes the where clause of the query
func (q *BlobQuery) Where(where string, args ...interface{}) *BlobQuery {
	q.whereQuery = where
	q.args = args //TODO: check if already have args, then add instead of set.
	return q
}

// First returns the first item found by the query
func (q *BlobQuery) First(ctx context.Context) (*Blob, error) {
	return q.driver.queryFirst(ctx, q)
}

// Slice returns a slice with all the matched item (warning: can use lots of memory)
func (q *BlobQuery) Slice(ctx context.Context, expectedSliceSize int) ([]*Blob, error) {
	return q.driver.querySlice(ctx, q, expectedSliceSize)
}

// Each returns a slice with all the matched item (warning: can use lots of memory)
func (q *BlobQuery) Each(ctx context.Context, reuseItem bool, action func(*Blob) error) error {
	return q.driver.queryEach(ctx, q, reuseItem, action)
}

type blobsDriver interface {
	insert(ctx context.Context, id []byte, _type int32, data []byte) (*Blob, error)
	load(ctx context.Context, query string, args ...interface{}) (*Blob, error)
	delete(ctx context.Context, query string, args ...interface{}) error
	save(ctx context.Context, blob *Blob) error
	//scan(blob *Blob, rows *sql.Rows) error
	queryFirst(ctx context.Context, query *BlobQuery) (*Blob, error)
	querySlice(ctx context.Context, query *BlobQuery, expectedSliceSize int) ([]*Blob, error)
	queryEach(ctx context.Context, query *BlobQuery, reuseItem bool, action func(*Blob) error) error
	loadByID(ctx context.Context, id []byte) (*Blob, error)
	findByID(id []byte) *BlobQuery
	deleteByID(ctx context.Context, id []byte) error
	loadByIDAndType(ctx context.Context, id []byte, _type int32) (*Blob, error)
	findByIDAndType(id []byte, _type int32) *BlobQuery
	deleteByIDAndType(ctx context.Context, id []byte, _type int32) error
}

// -------- cassandra --------

type blobsCassandraDriver struct {
	db *gocql.Session
}

func (d *blobsCassandraDriver) insert(ctx context.Context, id []byte, _type int32, data []byte) (*Blob, error) {
	return d.insertTTL(ctx, 0, id, _type, data)
}

// InsertTTL creates a record in the Cassandra Blobs table with the given TTL value
func (t BlobsTable) InsertTTL(ctx context.Context, ttl int64, id []byte, _type int32, data []byte) (*Blob, error) {
	return t.driver.(*blobsCassandraDriver).insertTTL(ctx, ttl, id, _type, data)
}

func (d *blobsCassandraDriver) insertTTL(ctx context.Context, ttl int64, id []byte, _type int32, data []byte) (*Blob, error) {
	if ctx == nil {
		ctx = context.Background()
	}

	cql := "insert into Blobs(id, type, data) values (?, ?, ?)"
	if ttl > 0 {
		cql += " USING TTL " + strconv.FormatInt(ttl, 10)
	}
	args := []interface{}{id, _type, data}
	ctx, done := logkit.Operation(ctx, "cassandra.cql", logkit.Stringer("cql", cqlStringer{cql: cql, args: args}))
	defer done()

	err := d.db.Query(cql, args...).Exec()

	if err != nil {
		return nil, logkit.Error(ctx, "CQL Error", logkit.Stringer("cql", cqlStringer{cql: cql, args: args}), logkit.Err(err))
	}

	return &Blob{
		driver: d,

		ID:   id,
		Type: _type,
		Data: data,

		loadID:   id,
		loadType: _type,
		loadData: data,
	}, nil
}

func (d *blobsCassandraDriver) load(ctx context.Context, query string, args ...interface{}) (*Blob, error) {
	if ctx == nil {
		ctx = context.Background()
	}
	cql := "SELECT id, type, data from Blobs where " + query
	ctx, done := logkit.Operation(ctx, "cassandra.cql", logkit.Stringer("cql", cqlStringer{cql: cql, args: args}))
	defer done()

	item := &Blob{}
	if err := d.db.Query(cql, args...).Scan(&item.ID, &item.Type, &item.Data); err != nil {
		if err == gocql.ErrNotFound {
			return nil, nil
		}
		return nil, logkit.Error(ctx, "CQL Error", logkit.Stringer("cql", cqlStringer{cql: cql, args: args}), logkit.Err(err))
	}
	item.driver = d
	item.loadID = item.ID
	item.loadType = item.Type
	item.loadData = item.Data

	return item, nil
}

func (d *blobsCassandraDriver) delete(ctx context.Context, query string, args ...interface{}) error {
	if ctx == nil {
		ctx = context.Background()
	}
	cql := "delete from Blobs where " + query
	ctx, done := logkit.Operation(ctx, "cassandra.cql", logkit.Stringer("cql", cqlStringer{cql: cql, args: args}))
	defer done()

	if err := d.db.Query(cql, args...).Exec(); err != nil {
		return logkit.Error(ctx, "CQL Error", logkit.Stringer("cql", cqlStringer{cql: cql, args: args}), logkit.Err(err))
	}
	return nil
}

func cassandraScanBlob(d *blobsCassandraDriver, iter *gocql.Iter, item *Blob) bool {
	result := iter.Scan(&item.ID, &item.Type, &item.Data)
	item.driver = d
	item.loadID = item.ID
	item.loadType = item.Type
	item.loadData = item.Data

	return result
	/*err := rows.Scan()
	if err != nil {
		return err
	}
	return nil
	return nil
	*/
}

func (d *blobsCassandraDriver) save(ctx context.Context, item *Blob) error {
	cql, args := getSaveBlobCQL(item)

	// nothings changed
	if cql == "" {
		return nil
	}

	// execute
	ctx, done := logkit.Operation(ctx, "cassandra.cql", logkit.Stringer("cql", cqlStringer{cql: cql, args: args}))
	defer done()

	if err := d.db.Query(cql, args...).Exec(); err != nil {
		return logkit.Error(ctx, "CQL Error", logkit.Stringer("cql", cqlStringer{cql: cql, args: args}), logkit.Err(err))
	}

	// reset load variables.
	item.resetLoadVars()
	return nil
}

func getSaveBlobCQL(item *Blob) (string, []interface{}) {
	var sb bytes.Buffer
	sb.WriteString("update Blobs set ")
	args := make([]interface{}, 0, 3)

	if !bytes.Equal(item.ID, item.loadID) {
		if len(args) != 0 {
			sb.WriteString(", ")
		}
		args = append(args, item.ID)
		sb.WriteString("id=?")
		//sb.WriteString(strconv.FormatInt(int64(len(args)), 10))
	}

	if item.Type != item.loadType {
		if len(args) != 0 {
			sb.WriteString(", ")
		}
		args = append(args, item.Type)
		sb.WriteString("type=?")
		//sb.WriteString(strconv.FormatInt(int64(len(args)), 10))
	}

	if !bytes.Equal(item.Data, item.loadData) {
		if len(args) != 0 {
			sb.WriteString(", ")
		}
		args = append(args, item.Data)
		sb.WriteString("data=?")
		//sb.WriteString(strconv.FormatInt(int64(len(args)), 10))
	}

	if len(args) == 0 {
		return "", args
	}

	// where clause
	sb.WriteString(" where  id=?  and  type=?")

	args = append(args, item.ID)
	args = append(args, item.Type)

	return sb.String(), args
}

func (i *Blob) resetLoadVars() {
	i.loadID = i.ID
	i.loadType = i.Type
	i.loadData = i.Data

}

func (d *blobsCassandraDriver) queryFirst(ctx context.Context, query *BlobQuery) (*Blob, error) {
	cql := d.queryCQL(query)
	ctx, done := logkit.Operation(ctx, "cassandra.cql", logkit.Stringer("cql", cqlStringer{cql: cql, args: query.args}))
	defer done()

	item := &Blob{}
	if err := d.db.Query(cql, query.args...).Scan(&item.ID, &item.Type, &item.Data); err != nil {
		if err == gocql.ErrNotFound {
			return nil, nil
		}
		return nil, logkit.Error(ctx, "CQL Error", logkit.Stringer("cql", cqlStringer{cql: cql, args: query.args}), logkit.Err(err))
	}
	item.driver = d
	item.loadID = item.ID
	item.loadType = item.Type
	item.loadData = item.Data

	return item, nil
}

func (d *blobsCassandraDriver) querySlice(ctx context.Context, query *BlobQuery, expectedSliceSize int) ([]*Blob, error) {
	cql := d.queryCQL(query)
	ctx, done := logkit.Operation(ctx, "cassandra.cql", logkit.Stringer("cql", cqlStringer{cql: cql, args: query.args}))
	defer done()

	iter := d.db.Query(cql, query.args...).Iter()
	result := make([]*Blob, 0, expectedSliceSize)
	for {
		item := &Blob{}
		if iter.Scan(&item.ID, &item.Type, &item.Data) {
			item.driver = d
			item.loadID = item.ID
			item.loadType = item.Type
			item.loadData = item.Data

			result = append(result, item)
		} else {
			break
		}
	}
	if err := iter.Close(); err != nil {
		return nil, logkit.Error(ctx, "CQL Error", logkit.Stringer("cql", cqlStringer{cql: cql, args: query.args}), logkit.Err(err))
	}
	return result, nil
}

func (d *blobsCassandraDriver) queryEach(ctx context.Context, query *BlobQuery, reuseItem bool, action func(*Blob) error) error {
	cql := d.queryCQL(query)
	ctx, done := logkit.Operation(ctx, "cassandra.cql", logkit.Stringer("cql", cqlStringer{cql: cql, args: query.args}))
	defer done()

	iter := d.db.Query(cql, query.args...).Iter()
	var item *Blob
	for {
		if item == nil || !reuseItem {
			item = &Blob{}
			item.driver = d
		}

		if iter.Scan(&item.ID, &item.Type, &item.Data) {
			item.loadID = item.ID
			item.loadType = item.Type
			item.loadData = item.Data

			if err := action(item); err != nil {
				iter.Close()
				return err
			}
		} else {
			break
		}

	}
	if err := iter.Close(); err != nil {
		return logkit.Error(ctx, "CQL Error", logkit.Stringer("cql", cqlStringer{cql: cql, args: query.args}), logkit.Err(err))
	}

	return nil
}

func (d *blobsCassandraDriver) queryCQL(query *BlobQuery) string {
	var sb bytes.Buffer

	sb.WriteString("Select id, type, data from Blobs")

	if query.whereQuery != "" {
		sb.WriteString(" WHERE ")
		sb.WriteString(query.whereQuery)
	}
	if len(query.orderColumns) > 0 {
		sb.WriteString(" ORDER BY ")
		for i, col := range query.orderColumns {
			if i > 0 {
				sb.WriteString(", ")
			}
			sb.WriteString(col)
			if query.orderDirection[i] == 1 {
				sb.WriteString(" ASC")
			} else {
				sb.WriteString(" DESC")
			}
		}

	}
	if query.limit > 0 {
		sb.WriteString(" LIMIT ")
		sb.WriteString(strconv.FormatInt(int64(query.limit), 10))
	}

	return sb.String()
}

func (d *blobsCassandraDriver) loadByID(ctx context.Context, id []byte) (*Blob, error) {
	return d.load(ctx, "id=?", id)
}

func (d *blobsCassandraDriver) findByID(id []byte) *BlobQuery {
	q := &BlobQuery{driver: d}
	q.Where("id=?", id)
	return q
}

func (d *blobsCassandraDriver) deleteByID(ctx context.Context, id []byte) error {
	return d.delete(ctx, "id=?", id)
}

func (d *blobsCassandraDriver) loadByIDAndType(ctx context.Context, id []byte, _type int32) (*Blob, error) {
	return d.load(ctx, "id=? and type=?", id, _type)
}

func (d *blobsCassandraDriver) findByIDAndType(id []byte, _type int32) *BlobQuery {
	q := &BlobQuery{driver: d}
	q.Where("id=? and type=?", id, _type)
	return q
}

func (d *blobsCassandraDriver) deleteByIDAndType(ctx context.Context, id []byte, _type int32) error {
	return d.delete(ctx, "id=? and type=?", id, _type)
}