<commit_before>package main

import (
	"encoding/json"
	"fmt"
	"strings"

	"gopkg.in/jackc/pgx.v2"

	"github.com/golang/groupcache"

	"github.com/LeKovr/dbrpc/workman"
	"github.com/LeKovr/go-base/logger"
)

// TableRow holds query result row
type TableRow map[string]interface{}

// TableRows holds slice of query result rows
type TableRows []TableRow

// -----------------------------------------------------------------------------

// Processor gets value from cache and converts it into Result struct
func cacheFetcher(log *logger.Log, cacheGroup *groupcache.Group) workman.WorkerFunc {
	// https://github.com/capotej/groupcache-db-experiment
	return func(payload string) workman.Result {
		var data []byte
		log.Debugf("asked for %s from groupcache", payload)
		err := cacheGroup.Get(nil, payload,
			groupcache.AllocatingByteSliceSink(&data))
		var res workman.Result
		if err != nil {
			res = workman.Result{Success: false, Error: err.Error()}
		} else if len(data) == 0 {
			res = workman.Result{Success: false, Error: "This internal error must be catched earlier. Please contact vendor"}
		} else {
			d := data[1:]
			raw := json.RawMessage(d)
			ok := data[0] == 1
			if ok { // First byte stores success state (1: true, 0: false)
				res = workman.Result{Success: ok, Result: &raw}
			} else {
				res = workman.Result{Success: ok, Error: &raw}
			}
		}
		return res
	}
}

// -----------------------------------------------------------------------------

func dbFetcher(cfg *AplFlags, log *logger.Log, db *pgx.ConnPool) groupcache.GetterFunc {
	return func(ctx groupcache.Context, key string, dest groupcache.Sink) error {
		log.Printf("asking for %s from dbserver", key)

		var isOk byte = 1 // status: success

		data, err := dbQuery(cfg, log, db, key)
		if err != nil {
			log.Warnf("Query for key %s error: %+v", key, err)
			isOk = 0
			data, _ = json.Marshal(err.Error())
		}

		result := []byte{isOk}
		result = append(result, data...)

		dd := result[1:]
		log.Debugf("Save data: %s", dd)
		dest.SetBytes([]byte(result))
		return nil
	}
}

// -----------------------------------------------------------------------------

// FuncDef holds function definition
type FuncDef struct {
	NspName string  // function namespace
	ProName string  // function name
	Permit  *string // permit code
	MaxAge  int     // cache max age
	IsRO    bool    // function is read-only (not volatile)
}

// FuncMap holds map of function definitions
type FuncMap map[string]FuncDef

// -----------------------------------------------------------------------------

func indexFetcher(cfg *AplFlags, log *logger.Log, db *pgx.ConnPool) (index *FuncMap, err error) {
	var rows *pgx.Rows

	q := fmt.Sprintf("select code, nspname, proname, permit_code, max_age, is_ro from %s()", cfg.ArgIndexFunc)
	log.Debugf("Query: %s", q)

	rows, err = db.Query(q)
	if err != nil {
		return
	}

	ind := FuncMap{}
	for rows.Next() {
		var fmr FuncDef
		var code string
		err = rows.Scan(&code, &fmr.NspName, &fmr.ProName, &fmr.Permit, &fmr.MaxAge, &fmr.IsRO)
		if err != nil {
			log.Warnf("Value fetch error: %s", err.Error())
			return
		}
		ind[code] = fmr
	}
	// Any errors encountered by rows.Next or rows.Scan will be returned here
	if rows.Err() != nil {
		err = rows.Err()
		return
	}
	log.Debugf("Index: %+v", ind)
	index = &ind
	return
}

// -----------------------------------------------------------------------------

func dbQuery(cfg *AplFlags, log *logger.Log, db *pgx.ConnPool, key string) (data []byte, err error) {
	var rows *pgx.Rows
	var q string
	var vals []interface{}
	var table *TableRows
	inTx := cfg.BeginFunc != "" // call in transaction
	var lang *string
	var tz *string
	if strings.HasPrefix(key, "[") {
		var args []interface{}
		err = json.Unmarshal([]byte(key), &args)<commit_msg>removed permit field from index call<commit_after>package main

import (
	"encoding/json"
	"fmt"
	"strings"

	"gopkg.in/jackc/pgx.v2"

	"github.com/golang/groupcache"

	"github.com/LeKovr/dbrpc/workman"
	"github.com/LeKovr/go-base/logger"
)

// TableRow holds query result row
type TableRow map[string]interface{}

// TableRows holds slice of query result rows
type TableRows []TableRow

// -----------------------------------------------------------------------------

// Processor gets value from cache and converts it into Result struct
func cacheFetcher(log *logger.Log, cacheGroup *groupcache.Group) workman.WorkerFunc {
	// https://github.com/capotej/groupcache-db-experiment
	return func(payload string) workman.Result {
		var data []byte
		log.Debugf("asked for %s from groupcache", payload)
		err := cacheGroup.Get(nil, payload,
			groupcache.AllocatingByteSliceSink(&data))
		var res workman.Result
		if err != nil {
			res = workman.Result{Success: false, Error: err.Error()}
		} else if len(data) == 0 {
			res = workman.Result{Success: false, Error: "This internal error must be catched earlier. Please contact vendor"}
		} else {
			d := data[1:]
			raw := json.RawMessage(d)
			ok := data[0] == 1
			if ok { // First byte stores success state (1: true, 0: false)
				res = workman.Result{Success: ok, Result: &raw}
			} else {
				res = workman.Result{Success: ok, Error: &raw}
			}
		}
		return res
	}
}

// -----------------------------------------------------------------------------

func dbFetcher(cfg *AplFlags, log *logger.Log, db *pgx.ConnPool) groupcache.GetterFunc {
	return func(ctx groupcache.Context, key string, dest groupcache.Sink) error {
		log.Printf("asking for %s from dbserver", key)

		var isOk byte = 1 // status: success

		data, err := dbQuery(cfg, log, db, key)
		if err != nil {
			log.Warnf("Query for key %s error: %+v", key, err)
			isOk = 0
			data, _ = json.Marshal(err.Error())
		}

		result := []byte{isOk}
		result = append(result, data...)

		dd := result[1:]
		log.Debugf("Save data: %s", dd)
		dest.SetBytes([]byte(result))
		return nil
	}
}

// -----------------------------------------------------------------------------

// FuncDef holds function definition
type FuncDef struct {
	NspName string // function namespace
	ProName string // function name
	//	Permit  *string // permit code
	MaxAge int  // cache max age
	IsRO   bool // function is read-only (not volatile)
}

// FuncMap holds map of function definitions
type FuncMap map[string]FuncDef

// -----------------------------------------------------------------------------

func indexFetcher(cfg *AplFlags, log *logger.Log, db *pgx.ConnPool) (index *FuncMap, err error) {
	var rows *pgx.Rows

	// q := fmt.Sprintf("select code, nspname, proname, permit_code, max_age, is_ro from %s()", cfg.ArgIndexFunc)
	q := fmt.Sprintf("select code, nspname, proname, max_age, is_ro from %s()", cfg.ArgIndexFunc)
	log.Debugf("Query: %s", q)

	rows, err = db.Query(q)
	if err != nil {
		return
	}

	ind := FuncMap{}
	for rows.Next() {
		var fmr FuncDef
		var code string
		// err = rows.Scan(&code, &fmr.NspName, &fmr.ProName, &fmr.Permit, &fmr.MaxAge, &fmr.IsRO)
		err = rows.Scan(&code, &fmr.NspName, &fmr.ProName, &fmr.MaxAge, &fmr.IsRO)
		if err != nil {
			log.Warnf("Value fetch error: %s", err.Error())
			return
		}
		ind[code] = fmr
	}
	// Any errors encountered by rows.Next or rows.Scan will be returned here
	if rows.Err() != nil {
		err = rows.Err()
		return
	}
	log.Debugf("Index: %+v", ind)
	index = &ind
	return
}

// -----------------------------------------------------------------------------

func dbQuery(cfg *AplFlags, log *logger.Log, db *pgx.ConnPool, key string) (data []byte, err error) {
	var rows *pgx.Rows
	var q string
	var vals []interface{}
	var table *TableRows
	inTx := cfg.BeginFunc != "" // call in transaction
	var lang *string
	var tz *string
	if strings.HasPrefix(key, "[") {
		var args []interface{}
		err = json.Unmarshal([]byte(key), &args)
{'fraction_non_alphanumeric': 0.2268983268983269, 'fraction_numerical': 0.002574002574002574, 'mean_word_length': 2.3366251610133104, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1002877', 'n_tokens_mistral': 2701, 'n_tokens_neox': 2538, 'n_words': 734}