func (s *sqlStore) rowsToRecords(rows *sql.Rows) ([]*store.Record, error) {
	var records []*store.Record
	var timehelper pq.NullTime
	for rows.Next() {
		record := &store.Record{}
		metadata := make(Metadata)
		if err := rows.Scan(&record.Key, &record.Value, &metadata, &timehelper); err != nil {
			return records, err
		}
		record.Metadata = toMetadata(&metadata)
		if timehelper.Valid {
			if timehelper.Time.Before(time.Now()) {
				go s.Delete(record.Key)
			} else {
				record.Expiry = time.Until(timehelper.Time)
				records = append(records, record)
			}
		} else {
			records = append(records, record)
		}
	}
	return records, nil
}
rowsToRecords converts from sql.Rows to  []*store.Record. If a record has expired it will issue a delete in a separate goroutine
package postgres
import (
	"database/sql"
	"database/sql/driver"
	"fmt"
	"net"
	"net/url"
	"regexp"
	"strings"
	"sync"
	"syscall"
	"time"
	"github.com/lib/pq"
	"github.com/micro/micro/v3/service/logger"
	"github.com/micro/micro/v3/service/store"
	"github.com/pkg/errors"
)
var (
	DefaultDatabase = "micro"
	DefaultTable    = "micro"
	ErrNoConnection = errors.New("Database connection not initialised")
)
var (
	re = regexp.MustCompile("[^a-zA-Z0-9]+")
	orderAsc  = "ORDER BY key ASC"
	orderDesc = "ORDER BY key DESC"
	statements = map[string]string{
		"list":          "SELECT key, value, metadata, expiry FROM %s.%s WHERE key LIKE $1 ORDER BY key ASC LIMIT $2 OFFSET $3;",
		"read":          "SELECT key, value, metadata, expiry FROM %s.%s WHERE key = $1;",
		"readMany":      "SELECT key, value, metadata, expiry FROM %s.%s WHERE key LIKE $1 ORDER BY key ASC;",
		"readOffset":    "SELECT key, value, metadata, expiry FROM %s.%s WHERE key LIKE $1 ORDER BY key ASC LIMIT $2 OFFSET $3;",
		"write":         "INSERT INTO %s.%s(key, value, metadata, expiry) VALUES ($1, $2::bytea, $3, $4) ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, metadata = EXCLUDED.metadata, expiry = EXCLUDED.expiry;",
		"delete":        "DELETE FROM %s.%s WHERE key = $1;",
		"deleteExpired": "DELETE FROM %s.%s WHERE expiry < now();",
		"showTables":    "SELECT schemaname, tablename FROM pg_catalog.pg_tables WHERE schemaname != 'pg_catalog' AND schemaname != 'information_schema';",
	}
)
type sqlStore struct {
	options store.Options
	dbConn  *sql.DB
	sync.RWMutex
	databases map[string]bool
}
func (s *sqlStore) getDB(database, table string) (string, string) {
	if len(database) == 0 {
		if len(s.options.Database) > 0 {
			database = s.options.Database
		} else {
			database = DefaultDatabase
		}
	}
	if len(table) == 0 {
		if len(s.options.Table) > 0 {
			table = s.options.Table
		} else {
			table = DefaultTable
		}
	}
	database = re.ReplaceAllString(database, "_")
	table = re.ReplaceAllString(table, "_")
	return database, table
}
func (s *sqlStore) createDB(database, table string) error {
	database, table = s.getDB(database, table)
	s.Lock()
	defer s.Unlock()
	if _, ok := s.databases[database+":"+table]; ok {
		return nil
	}
	if err := s.initDB(database, table); err != nil {
		return err
	}
	s.databases[database+":"+table] = true
	return nil
}
func (s *sqlStore) db() (*sql.DB, error) {
	if s.dbConn == nil {
		return nil, ErrNoConnection
	}
	if err := s.dbConn.Ping(); err != nil {
		if !isBadConnError(err) {
			return nil, err
		}
		logger.Errorf("Error with DB connection, will reconfigure: %s", err)
		if err := s.configure(); err != nil {
			logger.Errorf("Error while reconfiguring client: %s", err)
			return nil, err
		}
	}
	return s.dbConn, nil
}
func isBadConnError(err error) bool {
	if err == nil {
		return false
	}
	if err == driver.ErrBadConn {
		return true
	}
	if strings.Contains(err.Error(), syscall.ECONNRESET.Error()) {
		return true
	}
	switch t := err.(type) {
	case syscall.Errno:
		return t == syscall.ECONNRESET || t == syscall.ECONNABORTED || t == syscall.ECONNREFUSED
	case *net.OpError:
		return !t.Temporary()
	case net.Error:
		return !t.Temporary()
	}
	return false
}
func (s *sqlStore) initDB(database, table string) error {
	db, err := s.db()
	if err != nil {
		return err
	}
	_, err = db.Exec(fmt.Sprintf("CREATE DATABASE %s;", database))
	if err != nil && !strings.Contains(err.Error(), "already exists") {
		return err
	}
	var version string
	if err = db.QueryRow("select version()").Scan(&version); err == nil {
		if strings.Contains(version, "PostgreSQL") {
			_, err = db.Exec(fmt.Sprintf("CREATE SCHEMA IF NOT EXISTS %s;", database))
			if err != nil {
				return err
			}
		}
	}
	_, err = db.Exec(fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s.%s
	(
		key text NOT NULL,
		value bytea,
		metadata JSONB,
		expiry timestamp with time zone,
		CONSTRAINT %s_pkey PRIMARY KEY (key)
	);`, database, table, table))
	if err != nil {
		return errors.Wrap(err, "Couldn't create table")
	}
	_, err = db.Exec(fmt.Sprintf(`CREATE INDEX IF NOT EXISTS "%s" ON %s.%s USING btree ("key");`, "key_index_"+table, database, table))
	if err != nil {
		return err
	}
	_, err = db.Exec(fmt.Sprintf(`CREATE INDEX IF NOT EXISTS "%s" ON %s.%s USING GIN ("metadata");`, "metadata_index_"+table, database, table))
	if err != nil {
		return err
	}
	return nil
}
func (s *sqlStore) configure() error {
	if len(s.options.Nodes) == 0 {
		s.options.Nodes = []string{"postgresql://root@localhost:26257?sslmode=disable"}
	}
	source := s.options.Nodes[0]
	_, err := url.Parse(source)
	if err != nil {
		if !strings.Contains(source, " ") {
			source = fmt.Sprintf("host=%s", source)
		}
	}
	db, err := sql.Open("postgres", source)
	if err != nil {
		return err
	}
	if err := db.Ping(); err != nil {
		return err
	}
	if s.dbConn != nil {
		s.dbConn.Close()
	}
	s.dbConn = db
	database, table := s.getDB(s.options.Database, s.options.Table)
	return s.initDB(database, table)
}
func (s *sqlStore) prepare(database, table, query string, order store.Order) (*sql.Stmt, error) {
	st, ok := statements[query]
	if !ok {
		return nil, errors.New("unsupported statement")
	}
	if order == store.OrderDesc {
		st = strings.Replace(st, orderAsc, orderDesc, 1)
	}
	database, table = s.getDB(database, table)
	q := fmt.Sprintf(st, database, table)
	db, err := s.db()
	if err != nil {
		return nil, err
	}
	stmt, err := db.Prepare(q)
	if err != nil {
		return nil, err
	}
	return stmt, nil
}
func (s *sqlStore) Close() error {
	if s.dbConn != nil {
		return s.dbConn.Close()
	}
	return nil
}
func (s *sqlStore) Init(opts ...store.Option) error {
	for _, o := range opts {
		o(&s.options)
	}
	return s.configure()
}
func (s *sqlStore) List(opts ...store.ListOption) ([]string, error) {
	options := store.ListOptions{
		Order: store.OrderAsc,
	}
	for _, o := range opts {
		o(&options)
	}
	if err := s.createDB(options.Database, options.Table); err != nil {
		return nil, err
	}
	limit := sql.NullInt32{}
	offset := 0
	pattern := "%"
	if options.Prefix != "" || options.Suffix != "" {
		if options.Prefix != "" {
			pattern = options.Prefix + pattern
		}
		if options.Suffix != "" {
			pattern = pattern + options.Suffix
		}
	}
	if options.Offset > 0 {
		offset = int(options.Offset)
	}
	if options.Limit > 0 {
		limit = sql.NullInt32{Int32: int32(options.Limit), Valid: true}
	}
	st, err := s.prepare(options.Database, options.Table, "list", options.Order)
	if err != nil {
		return nil, err
	}
	defer st.Close()
	rows, err := st.Query(pattern, limit, offset)
	if err != nil {
		if err == sql.ErrNoRows {
			return nil, nil
		}
		return nil, err
	}
	defer rows.Close()
	var keys []string
	records, err := s.rowsToRecords(rows)
	if err != nil {
		return nil, err
	}
	for _, k := range records {
		keys = append(keys, k.Key)
	}
	rowErr := rows.Close()
	if rowErr != nil {
		return keys, rowErr
	}
	if err := rows.Err(); err != nil {
		return keys, err
	}
	return keys, nil
}
func (s *sqlStore) rowToRecord(row *sql.Row) (*store.Record, error) {
	var timehelper pq.NullTime
	record := &store.Record{}
	metadata := make(Metadata)
	if err := row.Scan(&record.Key, &record.Value, &metadata, &timehelper); err != nil {
		if err == sql.ErrNoRows {
			return record, store.ErrNotFound
		}
		return nil, err
	}
	record.Metadata = toMetadata(&metadata)
	if timehelper.Valid {
		if timehelper.Time.Before(time.Now()) {
			go s.Delete(record.Key)
			return nil, store.ErrNotFound
		}
		record.Expiry = time.Until(timehelper.Time)
	}
	return record, nil
}