<commit_before>// Copyright 2020 <NAME>
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Original source: github.com/micro/go-plugins/v3/store/postgres/postgres.go

// Package postgres implements the postgres store
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"
)

// DefaultDatabase is the namespace that the sql store
// will use if no namespace is provided.
var (
	DefaultDatabase = "micro"
	DefaultTable    = "micro"
	ErrNoConnection = errors.New("Database connection not initialised")
)

var (
	re = regexp.MustCompile("[^a-zA-Z0-9]+")

	// alternative ordering
	orderAsc  = "ORDER BY key ASC"
	orderDesc = "ORDER BY key DESC"

	// the sql statements we prepare and use
	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;",
	}
)

type sqlStore struct {
	options store.Options
	dbConn  *sql.DB

	sync.RWMutex
	// known databases
	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
		}
	}

	// store.namespace must only contain letters, numbers and underscores
	database = re.ReplaceAllString(database, "_")
	table = re.ReplaceAllString(table, "_")

	return database, table
}

// createDB ensures that the DB and table have been created. It's used for lazy initialisation
// and will record which tables have been created to reduce calls to the DB
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
}

// db returns a valid connection to the DB
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
}

// isBadConnError returns true if the error is related to having a bad connection such that you need to reconnect
func isBadConnError(err error) bool {
	if err == nil {
		return false
	}
	if err == driver.ErrBadConn {
		return true
	}

	// heavy handed crude check for "connection reset by peer"
	if strings.Contains(err.Error(), syscall.ECONNRESET.Error()) {
		return true
	}

	// otherwise iterate through the error types
	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
	}
	// Create the namespace's database
	_, err = db.Exec(fmt.Sprintf("CREATE DATABASE IF NOT EXISTS %s;", database))
	if err != nil {
		return err
	}

	// Create a table for the namespace's prefix
	_, 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")
	}
<commit_msg>make create database postgres compliant<commit_after>// Copyright 2020 <NAME>
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Original source: github.com/micro/go-plugins/v3/store/postgres/postgres.go

// Package postgres implements the postgres store
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"
)

// DefaultDatabase is the namespace that the sql store
// will use if no namespace is provided.
var (
	DefaultDatabase = "micro"
	DefaultTable    = "micro"
	ErrNoConnection = errors.New("Database connection not initialised")
)

var (
	re = regexp.MustCompile("[^a-zA-Z0-9]+")

	// alternative ordering
	orderAsc  = "ORDER BY key ASC"
	orderDesc = "ORDER BY key DESC"

	// the sql statements we prepare and use
	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;",
	}
)

type sqlStore struct {
	options store.Options
	dbConn  *sql.DB

	sync.RWMutex
	// known databases
	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
		}
	}

	// store.namespace must only contain letters, numbers and underscores
	database = re.ReplaceAllString(database, "_")
	table = re.ReplaceAllString(table, "_")

	return database, table
}

// createDB ensures that the DB and table have been created. It's used for lazy initialisation
// and will record which tables have been created to reduce calls to the DB
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
}

// db returns a valid connection to the DB
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
}

// isBadConnError returns true if the error is related to having a bad connection such that you need to reconnect
func isBadConnError(err error) bool {
	if err == nil {
		return false
	}
	if err == driver.ErrBadConn {
		return true
	}

	// heavy handed crude check for "connection reset by peer"
	if strings.Contains(err.Error(), syscall.ECONNRESET.Error()) {
		return true
	}

	// otherwise iterate through the error types
	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
	}
	// Create the namespace's database
	_, err = db.Exec(fmt.Sprintf("CREATE DATABASE %s;", database))
	if err != nil && !strings.Contains(err.Error(), "already exists") {
		return err
	}

	// Create a table for the namespace's prefix
	_, 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")
	}

{'fraction_non_alphanumeric': 0.126037139470565, 'fraction_numerical': 0.005926511260371394, 'mean_word_length': 3.54443447037702, 'pattern_counts': {'":': 16, '<': 5, '<?xml version=': 0, '>': 9, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, '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': '1332677', 'n_tokens_mistral': 3591, 'n_tokens_neox': 3317, 'n_words': 1160}