<commit_before>package postgres

import (
	"database/sql"
	"encoding/json"
	"fmt"
	"sort"
	"strconv"
	"strings"
	"time"

	"regexp"

	"github.com/lfq7413/tomato/errs"
	"github.com/lfq7413/tomato/types"
	"github.com/lfq7413/tomato/utils"
	"github.com/lib/pq"
)

const postgresSchemaCollectionName = "_SCHEMA"

const postgresRelationDoesNotExistError = "42P01"
const postgresDuplicateRelationError = "42P07"
const postgresDuplicateColumnError = "42701"
const postgresUniqueIndexViolationError = "23505"
const postgresTransactionAbortedError = "25P02"

// PostgresAdapter postgres 数据库适配器
type PostgresAdapter struct {
	collectionPrefix string
	collectionList   []string
	db               *sql.DB
}

// NewPostgresAdapter ...
func NewPostgresAdapter(collectionPrefix string, db *sql.DB) *PostgresAdapter {
	return &PostgresAdapter{
		collectionPrefix: collectionPrefix,
		collectionList:   []string{},
		db:               db,
	}
}

// ensureSchemaCollectionExists 确保 _SCHEMA 表存在，不存在则创建表
func (p *PostgresAdapter) ensureSchemaCollectionExists() error {
	_, err := p.db.Exec(`CREATE TABLE IF NOT EXISTS "_SCHEMA" ( "className" varChar(120), "schema" jsonb, "isParseClass" bool, PRIMARY KEY ("className") )`)
	if err != nil {
		if e, ok := err.(*pq.Error); ok {
			if e.Code == postgresDuplicateRelationError || e.Code == postgresUniqueIndexViolationError {
				// _SCHEMA 表已经存在，已经由其他请求创建，忽略错误
				return nil
			}
		} else {
			return err
		}
	}
	return nil
}

// ClassExists 检测数据库中是否存在指定类
func (p *PostgresAdapter) ClassExists(name string) bool {
	var result bool
	err := p.db.QueryRow(`SELECT EXISTS (SELECT 1 FROM   information_schema.tables WHERE table_name = $1)`, name).Scan(&result)
	if err != nil {
		return false
	}
	return result
}

// SetClassLevelPermissions 设置类级别权限
func (p *PostgresAdapter) SetClassLevelPermissions(className string, CLPs types.M) error {
	err := p.ensureSchemaCollectionExists()
	if err != nil {
		return err
	}
	if CLPs == nil {
		CLPs = types.M{}
	}
	b, err := json.Marshal(CLPs)
	if err != nil {
		return err
	}

	qs := `UPDATE "_SCHEMA" SET "schema" = json_object_set_key("schema", $1::text, $2::jsonb) WHERE "className"=$3 `
	_, err = p.db.Exec(qs, "classLevelPermissions", string(b), className)
	if err != nil {
		return err
	}

	return nil
}

// CreateClass 创建类
func (p *PostgresAdapter) CreateClass(className string, schema types.M) (types.M, error) {
	if schema == nil {
		schema = types.M{}
	}
	schema["className"] = className
	b, err := json.Marshal(schema)
	if err != nil {
		return nil, err
	}

	err = p.createTable(className, schema)
	if err != nil {
		return nil, err
	}

	_, err = p.db.Exec(`INSERT INTO "_SCHEMA" ("className", "schema", "isParseClass") VALUES ($1, $2, $3)`, className, string(b), true)
	if err != nil {
		if e, ok := err.(*pq.Error); ok {
			if e.Code == postgresUniqueIndexViolationError {
				return nil, errs.E(errs.DuplicateValue, "Class "+className+" already exists.")
			}
		}
		return nil, err
	}

	return toParseSchema(schema), nil
}

// createTable 仅创建表，不加入 schema 中
func (p *PostgresAdapter) createTable(className string, schema types.M) error {
	if schema == nil {
		schema = types.M{}
	}
	valuesArray := types.S{}
	patternsArray := []string{}
	var fields types.M
	if f := utils.M(schema["fields"]); f != nil {
		fields = f
	}

	if className == "_User" {
		fields["_email_verify_token_expires_at"] = types.M{"type": "Date"}
		fields["_email_verify_token"] = types.M{"type": "String"}
		fields["_account_lockout_expires_at"] = types.M{"type": "Date"}
		fields["_failed_login_count"] = types.M{"type": "Number"}
		fields["_perishable_token"] = types.M{"type": "String"}
		fields["_perishable_token_expires_at"] = types.M{"type": "Date"}
		fields["_password_changed_at"] = types.M{"type": "Date"}
		fields["_password_history"] = types.M{"type": "Array"}
	}

	relations := []string{}

	for fieldName, t := range fields {
		parseType := utils.M(t)
		if parseType == nil {
			parseType = types.M{}
		}

		if utils.S(parseType["type"]) == "Relation" {
			relations = append(relations, fieldName)
			continue
		}

		if fieldName == "_rperm" || fieldName == "_wperm" {
			parseType["contents"] = types.M{"type": "String"}
		}

		valuesArray = append(valuesArray, fieldName)
		postgresType, err := parseTypeToPostgresType(parseType)
		if err != nil {
			return err
		}
		valuesArray = append(valuesArray, postgresType)

		patternsArray = append(patternsArray, `"%s" %s`)
		if fieldName == "objectId" {
			valuesArray = append(valuesArray, fieldName)
			patternsArray = append(patternsArray, `PRIMARY KEY ("%s")`)
		}
	}

	qs := `CREATE TABLE IF NOT EXISTS "%s" (` + strings.Join(patternsArray, ",") + `)`
	values := append(types.S{className}, valuesArray...)
	qs = fmt.Sprintf(qs, values...)

	err := p.ensureSchemaCollectionExists()
	if err != nil {
		return err
	}

	_, err = p.db.Exec(qs)
	if err != nil {
		if e, ok := err.(*pq.Error); ok {
			if e.Code == postgresDuplicateRelationError {
				// 表已经存在，已经由其他请求创建，忽略错误
			} else {
				return err
			}
		} else {
			return err
		}
	}

	// 创建 relation 表
	for _, fieldName := range relations {
		name := fmt.Sprintf(`_Join:%s:%s`, fieldName, className)
		_, err = p.db.Exec(fmt.Sprintf(`CREATE TABLE IF NOT EXISTS "%s" ("relatedId" varChar(120), "owningId" varChar(120), PRIMARY KEY("relatedId", "owningId") )`, name))
		if err != nil {
			return err
		}
	}

	return nil
}

// AddFieldIfNotExists 添加字段定义
func (p *PostgresAdapter) AddFieldIfNotExists(className, fieldName string, fieldType types.M) error {
	if fieldType == nil {
		fieldType = types.M{}
	}

	if utils.S(fieldType["type"]) != "Relation" {
		tp, err := parseTypeToPostgresType(fieldType)
		if err != nil {
			return err
		}
		qs := fmt.Sprintf(`ALTER TABLE "%s" ADD COLUMN "%s" %s`, className, fieldName, tp)
		_, err = p.db.Exec(qs)
		if err != nil {
			if e, ok := err.(*pq.Error); ok {
				if e.Code == postgresRelationDoesNotExistError {
					// TODO 添加默认字段
					_, ce := p.CreateClass(className, types.M{"fields": types.M{fieldName: fieldType}})
					if ce != nil {
						return ce
					}
				} else if e.Code == postgresDuplicateColumnError {
					// Column 已经存在，由其他请求创建
				} else {
					return err
				}
			} else {
				return err
			}
		}
	} else {
		name := fmt.Sprintf(`_Join:%s:%s`, fieldName, className)
		qs := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS "%s" ("relatedId" varChar(120), "owningId" varChar(120), PRIMARY KEY("relatedId", "owningId") )`, name)
		_, err := p.db.Exec(qs)
		if err != nil {
			return err
		}
	}

	qs := `SELECT "schema" FROM "_SCHEMA" WHERE "className" = $1`
	rows, err := p.db.Query(qs, className)
	if err != nil {
		return err
	}
	if rows.Next() {
		var sch types.M
		var v []byte
		err := rows.Scan(&v)
		if err != nil {
			return err
		}
		err = json.Unmarshal(v, &sch)
		if err != nil {
			return err
		}
		if sch == nil {
			sch = types.M{}
		}
		var fields types.M
		if v := utils.M(sch["fields"]); v != nil {
			fields = v
		} else {
			fields = types.M{}
		}
		if _, ok := fields[fieldName]; ok {
			// 当表不存在时，会进行新建表，所以也会走到这里，不再处理错误
			// Attempted to add a field that already exists
			return nil
		}
		fields[fieldName] = fieldType
		sch["fields"] = fields
		b, err := json.Marshal(sch)
		qs := `UPDATE "_SCHEMA" SET "schema"=$1 WHERE "className"=$2`
		_, err = p.db.Exec(qs, b, className)
		if err != nil {
			return err
		}
	}

	return nil
}

// DeleteClass 删除指定表
func (p *PostgresAdapter) DeleteClass(className string) (types.M, error) {
	qs := fmt.Sprintf(`DROP TABLE IF EXISTS "%s"`, className)
	_, err := p.db.Exec(qs)
	if err != nil {
		return nil, err
	}

	qs = `DELETE FROM "_SCHEMA" WHERE "className"=$1`
	_, err = p.db.Exec(qs, className)
	if err != nil {
		return nil, err
	}

	return types.M{}, nil
}

// DeleteAllClasses 删除所有表，仅用于测试
func (p *PostgresAdapter) DeleteAllClasses() error {
	qs := `SELECT "className","schema" FROM "_SCHEMA"`
	rows, err := p.db.Query(qs)
	if err != nil {
		if e, ok := err.(*pq.Error); ok && e.Code == postgresRelationDoesNotExistError {
			// _SCHEMA 不存在，则不删除
			return nil
		}
		return err
	}

	classNames := []string{}
	schemas := []types.M{}

	for rows.Next() {
		var clsName string
		var sch types.M
		var v []byte
		err := rows.Scan(&clsName, &v)
		if err != nil {
			return err
		}
		err = json.Unmarshal(v, &sch)
		if err != nil {
			return err
		}
		classNames = append(classNames, clsName)
		schemas = append(schemas, sch)
	}

	joins := []string{}
	for _, sch := range schemas {
		joins = append(joins, joinTablesForSchema(sch)...)
	}

	classes := []string{"_SCHEMA", "_PushStatus", "_JobStatus", "_Hooks", "_GlobalConfig"}
	classes = append(classes, classNames...)
	classes = append(classes, joins...)

	for _, name := range classes {
		qs = fmt.Sprintf(`DROP TABLE IF EXISTS "%s"`, name)
		p.db.Exec(qs)
	}

	return nil
}

// DeleteFields 删除字段
func (p *PostgresAdapter) DeleteFields(className string, schema types.M, fieldNames []string) error {
	if schema == nil {
		schema = types.M{}
	}

	fields := utils.M(schema["fields"])
	if fields == nil {
		fields = types.M{}
	}
	fldNames := types.S{}
	for _, fieldName := range fieldNames {
		field := utils.M(fields[fieldName])
		if field != nil && utils.S(field["type"]) == "Relation" {
			// 不处理 Relation 类型字段
		} else {
			fldNames = append(fldNames, fieldName)
		}
		delete(fields, fieldName)
	}
	schema["fields"] = fields

	values := append(types.S{className}, fldNames...)
	columnArray := []string{}
	for _ = range fldNames {
		columnArray = append(columnArray, `"%s"`)
	}
	columns := strings.Join(columnArray, ",")

	b, err := json.Marshal(schema)
	if err != nil {
		return err
	}
	qs := `UPDATE "_SCHEMA" SET "schema"=$1 WHERE "className"=$2`
	_, err = p.db.Exec(qs, b, className)
	if err != nil {
		return err
	}

	if len(values) > 1 {
		qs = fmt.Sprintf(`ALTER TABLE "%%s" DROP COLUMN %s`, columns)
		qs = fmt.Sprintf(qs, values...)
		_, err = p.db.Exec(qs)
		if err != nil {
			return err
		}
	}

	return nil
}

// CreateObject 创建对象
func (p *PostgresAdapter) CreateObject(className string, schema, object types.M) error {
	columnsArray := []string{}
	valuesArray := types.S{}
	geoPoints := types.M{}
	if schema == nil {
		schema = types.M{}
	}
	if len(object) == 0 {
		return nil
	}
	schema = toPostgresSchema(schema)
	object = handleDotFields(object)

	err := validateKeys(object)
	if err != nil {
		return err
	}

	// 预处理 authData 字段，避免在向 map 添加元素时造成的不稳定性
	for fieldName := range object {
		re := regexp.MustCompile(`^_auth_data_([a-zA-Z0-9_]+)$`)
		authDataMatch := re.FindStringSubmatch(fieldName)
		if authDataMatch != nil && len(authDataMatch) == 2 {
			provider := authDataMatch[1]
			authData := utils.M(object["authData"])
			if authData == nil {
				authData = types.M{}
			}
			authData[provider] = object[fieldName]
			delete(object, fieldName)
			object["authData"] = authData
		}
	}

	for fieldName := range object {
		columnsArray = append(columnsArray, fieldName)

		fields := utils.M(schema["fields"])
		if fields == nil {
			fields = types.M{}
		}
		if fields[fieldName] == nil && className == "_User" {
			if fieldName == "_email_verify_token" ||
				fieldName == "_failed_login_count" ||
				fieldName == "_perishable_token" {
				valuesArray = append(valuesArray, object[fieldName])
			}

			if fieldName == "_password_history" {
				b, err := json.Marshal(object[fieldName])
				if err != nil {
					return err
				}
				valuesArray = append(valuesArray, b)
			}

			if fieldName == "_email_verify_token_expires_at" ||
				fieldName == "_account_lockout_expires_at" ||
				fieldName == "_perishable_token_expires_at" ||
				fieldName == "_password_changed_at" {
				if v := utils.M(object[fieldName]); v != nil && utils.S(v["iso"]) != "" {
					valuesArray = append(valuesArray, v["iso"])
				} else {
					valuesArray = append(valuesArray, nil)
				}
			}

			continue
		}

		tp := utils.M(fields[fieldName])
		if tp == nil {
			tp = types.M{}
		}
		switch utils.S(tp["type"]) {
		case "Date":
			if v := utils.M(object[fieldName]); v != nil && utils.S(v["iso"]) != "" {
				valuesArray = append(valuesArray, v["iso"])
			} else {
				valuesArray = append(valuesArray, nil)
			}
		case "Pointer":
			if v := utils.M(object[fieldName]); v != nil && utils.S(v["objectId"]) != "" {
				valuesArray = append(valuesArray, v["objectId"])
			} else {
				valuesArray = append(valuesArray, "")
			}
		case "Array":
			b, err := json.Marshal(object[fieldName])
			if err != nil {
				return err
			}
			if fieldName == "_rperm" || fieldName == "_wperm" {
				// '[' => '{'
				if b[0] == 91 {
					b[0] = 123
				}
				// ']' => '}'
				if len(b) > 0 && b[len(b)-1] == 93 {
					b[len(b)-1] = 125
				}
			}
			valuesArray = append(valuesArray, b)
		case "Object":
			b, err := json.Marshal(object[fieldName])
			if err != nil {
				return err
			}
			valuesArray = append(valuesArray, b)
		case "String", "Number", "Boolean":
			valuesArray = append(valuesArray, object[fieldName])
		case "File":
			if v := utils.M(object[fieldName]); v != nil && utils.S(v["name"]) != "" {
				valuesArray = append(valuesArray, v["name"])
			} else {
				valuesArray = append(valuesArray, "")
			}
		case "GeoPoint":
			geoPoints[fieldName] = object[fieldName]
			columnsArray = columnsArray[:len(columnsArray)-1]
		default:
			return errs.E(errs.OtherCause, "Type "+utils.S(tp["type"])+" not supported yet")
		}

	}

	for key := range geoPoints {
		columnsArray = append(columnsArray, key)
	}

	initialValues := []string{}
	for index := range valuesArray {
		termination := ""
		fieldName := columnsArray[index]
		if fieldName == "_rperm" || fieldName == "_wperm" {
			termination = "::text[]"
		} else {
			fields := utils.M(schema["fields"])
			if fields == nil {
				fields = types.M{}
			}
			tp := utils.M(fields[fieldName])
			if tp == nil {
				tp = types.M{}
			}
			if utils.S(tp["type"]) == "Array" {
				termination = "::jsonb"
			}
		}
		initialValues = append(initialValues, fmt.Sprintf(`$%d%s`, index+1, termination))
	}

	geoPointsInjects := []string{}
	for _, v := range geoPoints {
		value := utils.M(v)
		if value == nil {
			value = types.M{}
		}
		valuesArray = append(valuesArray, value["longitude"], value["latitude"])
		l := len(valuesArray)
		geoPointsInjects = append(geoPointsInjects, fmt.Sprintf(`POINT($%d, $%d)`, l-1, l))
	}

	columnsPatternArray := []string{}
	for _, key := range columnsArray {
		columnsPatternArray = append(columnsPatternArray, fmt.Sprintf(`"%s"`, key))
	}
	columnsPattern := strings.Join(columnsPatternArray, ",")

	initialValues = append(initialValues, geoPointsInjects...)
	valuesPattern := strings.Join(initialValues, ",")

	qs := fmt.Sprintf(`INSERT INTO "%s" (%s) VALUES (%s)`, className, columnsPattern, valuesPattern)
	_, err = p.db.Exec(qs, valuesArray...)
	if err != nil {
		if e, ok := err.(*pq.Error); ok {
			if e.Code == postgresUniqueIndexViolationError {
				return errs.E(errs.DuplicateValue, "A duplicate value for a field with unique values was provided")
			}
		}
		return err
	}

	return nil
}

// GetAllClasses ...
func (p *PostgresAdapter) GetAllClasses() ([]types.M, error) {
	err := p.ensureSchemaCollectionExists()
	if err != nil {
		return nil, err
	}
	qs := `SELECT "className","schema" FROM "_SCHEMA"`
	rows, err := p.db.Query(qs)
	if err != nil {
		return nil, err
	}

	schemas := []types.M{}

	for rows.Next() {
		var clsName string
		var sch types.M
		var v []byte
		err := rows.Scan(&clsName, &v)
		if err != nil {
			return nil, err
		}
		err = json.Unmarshal(v, &sch)
		if err != nil {
			return nil, err
		}
		sch["className"] = clsName
		schemas = append(schemas, toParseSchema(sch))
	}

	return schemas, nil
}

// GetClass ...
func (p *PostgresAdapter) GetClass(className string) (types.M, error) {
	err := p.ensureSchemaCollectionExists()
	if err != nil {
		return nil, err
	}
	qs := `SELECT "schema" FROM "_SCHEMA" WHERE "className"=$1`
	rows, err := p.db.Query(qs, className)
	if err != nil {
		return nil, err
	}

	schema := types.M{}
	if rows.Next() {
		var v []byte
		err = rows.Scan(&v)
		if err != nil {
			return nil, err
		}
		err = json.Unmarshal(v, &schema)
		if err != nil {
			return nil, err
		}
	} else {
		return schema, nil
	}

	return toParseSchema(schema), nil
}

// DeleteObjectsByQuery 删除符合条件的所有对象
func (p *PostgresAdapter) DeleteObjectsByQuery(className string, schema, query types.M) error {
	where, err := buildWhereClause(schema, query, 1)
	if err != nil {
		return err
	}

	if len(query) == 0 {
		where.pattern = "TRUE"
	}

	qs := fmt.Sprintf(`WITH deleted AS (DELETE FROM "%s" WHERE %s RETURNING *) SELECT count(*) FROM deleted`, className, where.pattern)
	row := p.db.QueryRow(qs, where.values...)
	var count int
	err = row.Scan(&count)
	if err != nil {
		if e, ok := err.(*pq.Error); ok {
			// 表不存在返回空
			if e.Code == postgresRelationDoesNotExistError {
				return errs.E(errs.ObjectNotFound, "Object not found.")
			}
		}
		return err
	}

	if count == 0 {
		return errs.E(errs.ObjectNotFound, "Object not found.")
	}

	return nil
}

// Find ...
func (p *PostgresAdapter) Find(className string, schema, query, options types.M) ([]types.M, error) {
	if schema == nil {
		schema = types.M{}
	}
	if options == nil {
		options = types.M{}
	}

	var hasLimit bool
	var hasSkip bool
	if _, ok := options["limit"]; ok {
		hasLimit = true
	}
	if _, ok := options["skip"]; ok {
		hasSkip = true
	}

	values := types.S{}
	where, err := buildWhereClause(schema, query, 1)
	if err != nil {
		return nil, err
	}
	values = append(values, where.values...)

	var wherePattern string
	var limitPattern string
	var skipPattern string
	if where.pattern != "" {
		wherePattern = `WHERE ` + where.pattern
	}
	if hasLimit {
		limitPattern = fmt.Sprintf(`LIMIT $%d`, len(values)+1)
		values = append(values, options["limit"])
	}
	if hasSkip {
		skipPattern = fmt.Sprintf(`OFFSET $%d`, len(values)+1)
		values = append(values, options["skip"])
	}

	var sortPattern string
	if _, ok := options["sort"]; ok {
		if keys, ok := options["sort"].([]string); ok {
			postgresSort := []string{}
			for _, key := range keys {
				var postgresKey string
				if strings.HasPrefix(key, "-") {
					key = key[1:]
					postgresKey = fmt.Sprintf(`"%s" DESC`, key)
				} else {
					postgresKey = fmt.Sprintf(`"%s" ASC`, key)
				}
				postgresSort = append(postgresSort, postgresKey)
			}
			sorting := strings.Join(postgresSort, ",")
			if len(postgresSort) > 0 {
				sortPattern = fmt.Sprintf(`ORDER BY %s`, sorting)
			}
		}
	}
	if len(where.sorts) > 0 {
		sortPattern = fmt.Sprintf(`ORDER BY %s`, strings.Join(where.sorts, ","))
	}

	columns := "*"
	if _, ok := options["keys"]; ok {
		if keys, ok := options["keys"].([]string); ok {
			postgresKeys := []string{}
			for _, key := range keys {
				if key != "" {
					postgresKeys = append(postgresKeys, fmt.Sprintf(`"%s"`, key))
				}
			}
			if len(postgresKeys) > 0 {
				columns = strings.Join(postgresKeys, ",")
			}
		}
	}

	qs := fmt.Sprintf(`SELECT %s FROM "%s" %s %s %s %s`, columns, className, wherePattern, sortPattern, limitPattern, skipPattern)
	rows, err := p.db.Query(qs, values...)
	if err != nil {
		if e, ok := err.(*pq.Error); ok {
			// 表不存在返回空
			if e.Code == postgresRelationDoesNotExistError {
				return []types.M{}, nil
			}
		}
		return nil, err
	}

	fields := utils.M(schema["fields"])
	if fields == nil {
		fields = types.M{}
	}

	results := []types.M{}
	var resultColumns []string
	for rows.Next() {
		if resultColumns == nil {
			resultColumns, err = rows.Columns()
			if err != nil {
				return nil, err
			}
		}
		resultValues := []*interface{}{}
		values := types.S{}
		for i := 0; i < len(resultColumns); i++ {
			var v interface{}
			resultValues = append(resultValues, &v)
			values = append(values, &v)
		}
		err = rows.Scan(values...)
		if err != nil {
			return nil, err
		}
		object := types.M{}
		for i, field := range resultColumns {
			object[field] = *resultValues[i]
		}

		object, err = postgresObjectToParseObject(object, fields)
		if err != nil {
			return nil, err
		}

		results = append(results, object)
	}

	return results, nil
}

// Count ...
func (p *PostgresAdapter) Count(className string, schema, query types.M) (int, error) {
	where, err := buildWhereClause(schema, query, 1)
	if err != nil {
		return 0, err
	}

	wherePattern := ""
	if len(where.pattern) > 0 {
		wherePattern = `WHERE ` + where.pattern
	}

	qs := fmt.Sprintf(`SELECT count(*) FROM "%s" %s`, className, wherePattern)
	rows, err := p.db.Query(qs, where.values...)
	if err != nil {
		if e, ok := err.(*pq.Error); ok {
			if e.Code == postgresRelationDoesNotExistError {
				return 0, nil
			}
		}
		return 0, err
	}
	var count int
	if rows.Next() {
		err = rows.Scan(&count)
		if err != nil {
			return 0, nil
		}
	}

	return count, nil
}

// UpdateObjectsByQuery ...
func (p *PostgresAdapter) UpdateObjectsByQuery(className string, schema, query, update types.M) error {
	_, err := p.FindOneAndUpdate(className, schema, query, update)
	return err
}

// FindOneAndUpdate ...
func (p *PostgresAdapter) FindOneAndUpdate(className string, schema, query, update types.M) (types.M, error) {
	updatePatterns := []string{}
	values := types.S{}
	index := 1

	if schema == nil {
		schema = types.M{}
	}
	schema = toPostgresSchema(schema)

	fields := utils.M(schema["fields"])
	if fields == nil {
		fields = types.M{}
	}

	originalUpdate := utils.CopyMapM(update)
	update = handleDotFields(update)

	for fieldName, v := range update {
		re := regexp.MustCompile(`^_auth_data_([a-zA-Z0-9_]+)$`)
		authDataMatch := re.FindStringSubmatch(fieldName)
		if authDataMatch != nil && len(authDataMatch) == 2 {
			provider := authDataMatch[1]
			delete(update, fieldName)
			authData := utils.M(update["authData"])
			if authData == nil {
				authData = types.M{}
			}
			authData[provider] = v
			update["authData"] = authData
		}
	}

	for fieldName, fieldValue := range update {
		if fieldValue == nil {
			updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = NULL`, fieldName))
			continue
		}

		if fieldName == "authData" {
			generate := func(jsonb, key, value string) string {
				return fmt.Sprintf(`json_object_set_key(COALESCE(%s, '{}'::jsonb), %s, %s)::jsonb`, jsonb, key, value)
			}
			lastKey := fmt.Sprintf(`"%s"`, fieldName)
			authData := utils.M(fieldValue)
			if authData == nil {
				continue
			}
			for key, value := range authData {
				lastKey = generate(lastKey, fmt.Sprintf(`$%d::text`, index), fmt.Sprintf(`$%d::jsonb`, index+1))
				index = index + 2
				if value != nil {
					if v := utils.M(value); v != nil && utils.S(v["__op"]) == "Delete" {
						value = nil
					} else {
						b, err := json.Marshal(v)
						if err != nil {
							return nil, err
						}
						value = string(b)
					}
				}
				values = append(values, key, value)
			}
			updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = %s`, fieldName, lastKey))
			continue
		}

		if fieldName == "updatedAt" {
			updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = $%d`, fieldName, index))
			values = append(values, fieldValue)
			index = index + 1
			continue
		}

		switch fieldValue.(type) {
		case string, bool, float64, int:
			updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = $%d`, fieldName, index))
			values = append(values, fieldValue)
			index = index + 1
			continue
		case time.Time:
			updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = $%d`, fieldName, index))
			values = append(values, utils.TimetoString(fieldValue.(time.Time)))
			index = index + 1
			continue
		}

		if object := utils.M(fieldValue); object != nil {
			switch utils.S(object["__op"]) {
			case "Increment":
				updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = COALESCE("%s", 0) + $%d`, fieldName, fieldName, index))
				values = append(values, object["amount"])
				index = index + 1
				continue
			case "Add":
				updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = array_add(COALESCE("%s", '[]'::jsonb), $%d::jsonb)`, fieldName, fieldName, index))
				b, err := json.Marshal(object["objects"])
				if err != nil {
					return nil, err
				}
				values = append(values, string(b))
				index = index + 1
				continue
			case "Delete":
				updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = $%d`, fieldName, index))
				values = append(values, nil)
				index = index + 1
				continue
			case "Remove":
				updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = array_remove(COALESCE("%s", '[]'::jsonb), $%d::jsonb)`, fieldName, fieldName, index))
				b, err := json.Marshal(object["objects"])
				if err != nil {
					return nil, err
				}
				values = append(values, string(b))
				index = index + 1
				continue
			case "AddUnique":
				updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = array_add_unique(COALESCE("%s", '[]'::jsonb), $%d::jsonb)`, fieldName, fieldName, index))
				b, err := json.Marshal(object["objects"])
				if err != nil {
					return nil, err
				}
				values = append(values, string(b))
				index = index + 1
				continue
			}

			switch utils.S(object["__type"]) {
			case "Pointer":
				updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = $%d`, fieldName, index))
				values = append(values, object["objectId"])
				index = index + 1
				continue
			case "Date", "File":
				updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = $%d`, fieldName, index))
				values = append(values, toPostgresValue(object))
				index = index + 1
				continue
			case "GeoPoint":
				updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = POINT($%d, $%d)`, fieldName, index, index+1))
				values = append(values, object["longitude"], object["latitude"])
				index = index + 2
				continue
			case "Relation":
				continue
			}

			if tp := utils.M(fields[fieldName]); tp != nil && utils.S(tp["type"]) == "Object" {
				keysToDelete := []string{}
				for k, v := range originalUpdate {
					if o := utils.M(v); o != nil && utils.S(o["__op"]) == "Delete" {
						if keys := strings.Split(k, "."); len(keys) == 2 && keys[0] == fieldName {
							keysToDelete = append(keysToDelete, keys[1])
						}
					}
				}

				deletePatterns := ""
				for _, k := range keysToDelete {
					deletePatterns = deletePatterns + fmt.Sprintf(` - '%s'`, k)
				}

				updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = ( COALESCE("%s", '{}'::jsonb) %s || $%d::jsonb )`, fieldName, fieldName, deletePatterns, index))
				b, err := json.Marshal(object)
				if err != nil {
					return nil, err
				}
				values = append(values, string(b))
				index = index + 1
				continue
			}
		}

		if array := utils.A(fieldValue); array != nil {
			if tp := utils.M(fields[fieldName]); tp != nil && utils.S(tp["type"]) == "Array" {
				expectedType, err := parseTypeToPostgresType(tp)
				if err != nil {
					return nil, err
				}

				b, err := json.Marshal(fieldValue)
				if err != nil {
					return nil, err
				}

				if expectedType == "text[]" {
					// '[' => '{'
					if b[0] == 91 {
						b[0] = 123
					}
					// ']' => '}'
					if len(b) > 0 && b[len(b)-1] == 93 {
						b[len(b)-1] = 125
					}
					updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = $%d::text[]`, fieldName, index))
				} else {
					updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = $%d::jsonb`, fieldName, index))
				}

				values = append(values, string(b))
				index = index + 1
				continue
			}
		}

		b, _ := json.Marshal(fieldValue)
		return nil, errs.E(errs.OperationForbidden, "Postgres doesn't support update "+string(b)+" yet")
	}

	where, err := buildWhereClause(schema, query, index)
	if err != nil {
		return nil, err
	}
	values = append(values, where.values...)

	// TODO 需要添加限制，只更新一条，UpdateObjectsByQuery 时更新多条
	qs := fmt.Sprintf(`UPDATE "%s" SET %s WHERE %s RETURNING *`, className, strings.Join(updatePatterns, ","), where.pattern)
	rows, err := p.db.Query(qs, values...)
	if err != nil {
		return nil, err
	}<commit_msg>FindOneAndUpdate 转换表不存在的错误<commit_after>package postgres

import (
	"database/sql"
	"encoding/json"
	"fmt"
	"sort"
	"strconv"
	"strings"
	"time"

	"regexp"

	"github.com/lfq7413/tomato/errs"
	"github.com/lfq7413/tomato/types"
	"github.com/lfq7413/tomato/utils"
	"github.com/lib/pq"
)

const postgresSchemaCollectionName = "_SCHEMA"

const postgresRelationDoesNotExistError = "42P01"
const postgresDuplicateRelationError = "42P07"
const postgresDuplicateColumnError = "42701"
const postgresUniqueIndexViolationError = "23505"
const postgresTransactionAbortedError = "25P02"

// PostgresAdapter postgres 数据库适配器
type PostgresAdapter struct {
	collectionPrefix string
	collectionList   []string
	db               *sql.DB
}

// NewPostgresAdapter ...
func NewPostgresAdapter(collectionPrefix string, db *sql.DB) *PostgresAdapter {
	return &PostgresAdapter{
		collectionPrefix: collectionPrefix,
		collectionList:   []string{},
		db:               db,
	}
}

// ensureSchemaCollectionExists 确保 _SCHEMA 表存在，不存在则创建表
func (p *PostgresAdapter) ensureSchemaCollectionExists() error {
	_, err := p.db.Exec(`CREATE TABLE IF NOT EXISTS "_SCHEMA" ( "className" varChar(120), "schema" jsonb, "isParseClass" bool, PRIMARY KEY ("className") )`)
	if err != nil {
		if e, ok := err.(*pq.Error); ok {
			if e.Code == postgresDuplicateRelationError || e.Code == postgresUniqueIndexViolationError {
				// _SCHEMA 表已经存在，已经由其他请求创建，忽略错误
				return nil
			}
		} else {
			return err
		}
	}
	return nil
}

// ClassExists 检测数据库中是否存在指定类
func (p *PostgresAdapter) ClassExists(name string) bool {
	var result bool
	err := p.db.QueryRow(`SELECT EXISTS (SELECT 1 FROM   information_schema.tables WHERE table_name = $1)`, name).Scan(&result)
	if err != nil {
		return false
	}
	return result
}

// SetClassLevelPermissions 设置类级别权限
func (p *PostgresAdapter) SetClassLevelPermissions(className string, CLPs types.M) error {
	err := p.ensureSchemaCollectionExists()
	if err != nil {
		return err
	}
	if CLPs == nil {
		CLPs = types.M{}
	}
	b, err := json.Marshal(CLPs)
	if err != nil {
		return err
	}

	qs := `UPDATE "_SCHEMA" SET "schema" = json_object_set_key("schema", $1::text, $2::jsonb) WHERE "className"=$3 `
	_, err = p.db.Exec(qs, "classLevelPermissions", string(b), className)
	if err != nil {
		return err
	}

	return nil
}

// CreateClass 创建类
func (p *PostgresAdapter) CreateClass(className string, schema types.M) (types.M, error) {
	if schema == nil {
		schema = types.M{}
	}
	schema["className"] = className
	b, err := json.Marshal(schema)
	if err != nil {
		return nil, err
	}

	err = p.createTable(className, schema)
	if err != nil {
		return nil, err
	}

	_, err = p.db.Exec(`INSERT INTO "_SCHEMA" ("className", "schema", "isParseClass") VALUES ($1, $2, $3)`, className, string(b), true)
	if err != nil {
		if e, ok := err.(*pq.Error); ok {
			if e.Code == postgresUniqueIndexViolationError {
				return nil, errs.E(errs.DuplicateValue, "Class "+className+" already exists.")
			}
		}
		return nil, err
	}

	return toParseSchema(schema), nil
}

// createTable 仅创建表，不加入 schema 中
func (p *PostgresAdapter) createTable(className string, schema types.M) error {
	if schema == nil {
		schema = types.M{}
	}
	valuesArray := types.S{}
	patternsArray := []string{}
	var fields types.M
	if f := utils.M(schema["fields"]); f != nil {
		fields = f
	}

	if className == "_User" {
		fields["_email_verify_token_expires_at"] = types.M{"type": "Date"}
		fields["_email_verify_token"] = types.M{"type": "String"}
		fields["_account_lockout_expires_at"] = types.M{"type": "Date"}
		fields["_failed_login_count"] = types.M{"type": "Number"}
		fields["_perishable_token"] = types.M{"type": "String"}
		fields["_perishable_token_expires_at"] = types.M{"type": "Date"}
		fields["_password_changed_at"] = types.M{"type": "Date"}
		fields["_password_history"] = types.M{"type": "Array"}
	}

	relations := []string{}

	for fieldName, t := range fields {
		parseType := utils.M(t)
		if parseType == nil {
			parseType = types.M{}
		}

		if utils.S(parseType["type"]) == "Relation" {
			relations = append(relations, fieldName)
			continue
		}

		if fieldName == "_rperm" || fieldName == "_wperm" {
			parseType["contents"] = types.M{"type": "String"}
		}

		valuesArray = append(valuesArray, fieldName)
		postgresType, err := parseTypeToPostgresType(parseType)
		if err != nil {
			return err
		}
		valuesArray = append(valuesArray, postgresType)

		patternsArray = append(patternsArray, `"%s" %s`)
		if fieldName == "objectId" {
			valuesArray = append(valuesArray, fieldName)
			patternsArray = append(patternsArray, `PRIMARY KEY ("%s")`)
		}
	}

	qs := `CREATE TABLE IF NOT EXISTS "%s" (` + strings.Join(patternsArray, ",") + `)`
	values := append(types.S{className}, valuesArray...)
	qs = fmt.Sprintf(qs, values...)

	err := p.ensureSchemaCollectionExists()
	if err != nil {
		return err
	}

	_, err = p.db.Exec(qs)
	if err != nil {
		if e, ok := err.(*pq.Error); ok {
			if e.Code == postgresDuplicateRelationError {
				// 表已经存在，已经由其他请求创建，忽略错误
			} else {
				return err
			}
		} else {
			return err
		}
	}

	// 创建 relation 表
	for _, fieldName := range relations {
		name := fmt.Sprintf(`_Join:%s:%s`, fieldName, className)
		_, err = p.db.Exec(fmt.Sprintf(`CREATE TABLE IF NOT EXISTS "%s" ("relatedId" varChar(120), "owningId" varChar(120), PRIMARY KEY("relatedId", "owningId") )`, name))
		if err != nil {
			return err
		}
	}

	return nil
}

// AddFieldIfNotExists 添加字段定义
func (p *PostgresAdapter) AddFieldIfNotExists(className, fieldName string, fieldType types.M) error {
	if fieldType == nil {
		fieldType = types.M{}
	}

	if utils.S(fieldType["type"]) != "Relation" {
		tp, err := parseTypeToPostgresType(fieldType)
		if err != nil {
			return err
		}
		qs := fmt.Sprintf(`ALTER TABLE "%s" ADD COLUMN "%s" %s`, className, fieldName, tp)
		_, err = p.db.Exec(qs)
		if err != nil {
			if e, ok := err.(*pq.Error); ok {
				if e.Code == postgresRelationDoesNotExistError {
					// TODO 添加默认字段
					_, ce := p.CreateClass(className, types.M{"fields": types.M{fieldName: fieldType}})
					if ce != nil {
						return ce
					}
				} else if e.Code == postgresDuplicateColumnError {
					// Column 已经存在，由其他请求创建
				} else {
					return err
				}
			} else {
				return err
			}
		}
	} else {
		name := fmt.Sprintf(`_Join:%s:%s`, fieldName, className)
		qs := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS "%s" ("relatedId" varChar(120), "owningId" varChar(120), PRIMARY KEY("relatedId", "owningId") )`, name)
		_, err := p.db.Exec(qs)
		if err != nil {
			return err
		}
	}

	qs := `SELECT "schema" FROM "_SCHEMA" WHERE "className" = $1`
	rows, err := p.db.Query(qs, className)
	if err != nil {
		return err
	}
	if rows.Next() {
		var sch types.M
		var v []byte
		err := rows.Scan(&v)
		if err != nil {
			return err
		}
		err = json.Unmarshal(v, &sch)
		if err != nil {
			return err
		}
		if sch == nil {
			sch = types.M{}
		}
		var fields types.M
		if v := utils.M(sch["fields"]); v != nil {
			fields = v
		} else {
			fields = types.M{}
		}
		if _, ok := fields[fieldName]; ok {
			// 当表不存在时，会进行新建表，所以也会走到这里，不再处理错误
			// Attempted to add a field that already exists
			return nil
		}
		fields[fieldName] = fieldType
		sch["fields"] = fields
		b, err := json.Marshal(sch)
		qs := `UPDATE "_SCHEMA" SET "schema"=$1 WHERE "className"=$2`
		_, err = p.db.Exec(qs, b, className)
		if err != nil {
			return err
		}
	}

	return nil
}

// DeleteClass 删除指定表
func (p *PostgresAdapter) DeleteClass(className string) (types.M, error) {
	qs := fmt.Sprintf(`DROP TABLE IF EXISTS "%s"`, className)
	_, err := p.db.Exec(qs)
	if err != nil {
		return nil, err
	}

	qs = `DELETE FROM "_SCHEMA" WHERE "className"=$1`
	_, err = p.db.Exec(qs, className)
	if err != nil {
		return nil, err
	}

	return types.M{}, nil
}

// DeleteAllClasses 删除所有表，仅用于测试
func (p *PostgresAdapter) DeleteAllClasses() error {
	qs := `SELECT "className","schema" FROM "_SCHEMA"`
	rows, err := p.db.Query(qs)
	if err != nil {
		if e, ok := err.(*pq.Error); ok && e.Code == postgresRelationDoesNotExistError {
			// _SCHEMA 不存在，则不删除
			return nil
		}
		return err
	}

	classNames := []string{}
	schemas := []types.M{}

	for rows.Next() {
		var clsName string
		var sch types.M
		var v []byte
		err := rows.Scan(&clsName, &v)
		if err != nil {
			return err
		}
		err = json.Unmarshal(v, &sch)
		if err != nil {
			return err
		}
		classNames = append(classNames, clsName)
		schemas = append(schemas, sch)
	}

	joins := []string{}
	for _, sch := range schemas {
		joins = append(joins, joinTablesForSchema(sch)...)
	}

	classes := []string{"_SCHEMA", "_PushStatus", "_JobStatus", "_Hooks", "_GlobalConfig"}
	classes = append(classes, classNames...)
	classes = append(classes, joins...)

	for _, name := range classes {
		qs = fmt.Sprintf(`DROP TABLE IF EXISTS "%s"`, name)
		p.db.Exec(qs)
	}

	return nil
}

// DeleteFields 删除字段
func (p *PostgresAdapter) DeleteFields(className string, schema types.M, fieldNames []string) error {
	if schema == nil {
		schema = types.M{}
	}

	fields := utils.M(schema["fields"])
	if fields == nil {
		fields = types.M{}
	}
	fldNames := types.S{}
	for _, fieldName := range fieldNames {
		field := utils.M(fields[fieldName])
		if field != nil && utils.S(field["type"]) == "Relation" {
			// 不处理 Relation 类型字段
		} else {
			fldNames = append(fldNames, fieldName)
		}
		delete(fields, fieldName)
	}
	schema["fields"] = fields

	values := append(types.S{className}, fldNames...)
	columnArray := []string{}
	for _ = range fldNames {
		columnArray = append(columnArray, `"%s"`)
	}
	columns := strings.Join(columnArray, ",")

	b, err := json.Marshal(schema)
	if err != nil {
		return err
	}
	qs := `UPDATE "_SCHEMA" SET "schema"=$1 WHERE "className"=$2`
	_, err = p.db.Exec(qs, b, className)
	if err != nil {
		return err
	}

	if len(values) > 1 {
		qs = fmt.Sprintf(`ALTER TABLE "%%s" DROP COLUMN %s`, columns)
		qs = fmt.Sprintf(qs, values...)
		_, err = p.db.Exec(qs)
		if err != nil {
			return err
		}
	}

	return nil
}

// CreateObject 创建对象
func (p *PostgresAdapter) CreateObject(className string, schema, object types.M) error {
	columnsArray := []string{}
	valuesArray := types.S{}
	geoPoints := types.M{}
	if schema == nil {
		schema = types.M{}
	}
	if len(object) == 0 {
		return nil
	}
	schema = toPostgresSchema(schema)
	object = handleDotFields(object)

	err := validateKeys(object)
	if err != nil {
		return err
	}

	// 预处理 authData 字段，避免在遍历 map 并向其添加元素时造成的不稳定性
	for fieldName := range object {
		re := regexp.MustCompile(`^_auth_data_([a-zA-Z0-9_]+)$`)
		authDataMatch := re.FindStringSubmatch(fieldName)
		if authDataMatch != nil && len(authDataMatch) == 2 {
			provider := authDataMatch[1]
			authData := utils.M(object["authData"])
			if authData == nil {
				authData = types.M{}
			}
			authData[provider] = object[fieldName]
			delete(object, fieldName)
			object["authData"] = authData
		}
	}

	for fieldName := range object {
		columnsArray = append(columnsArray, fieldName)

		fields := utils.M(schema["fields"])
		if fields == nil {
			fields = types.M{}
		}
		if fields[fieldName] == nil && className == "_User" {
			if fieldName == "_email_verify_token" ||
				fieldName == "_failed_login_count" ||
				fieldName == "_perishable_token" {
				valuesArray = append(valuesArray, object[fieldName])
			}

			if fieldName == "_password_history" {
				b, err := json.Marshal(object[fieldName])
				if err != nil {
					return err
				}
				valuesArray = append(valuesArray, b)
			}

			if fieldName == "_email_verify_token_expires_at" ||
				fieldName == "_account_lockout_expires_at" ||
				fieldName == "_perishable_token_expires_at" ||
				fieldName == "_password_changed_at" {
				if v := utils.M(object[fieldName]); v != nil && utils.S(v["iso"]) != "" {
					valuesArray = append(valuesArray, v["iso"])
				} else {
					valuesArray = append(valuesArray, nil)
				}
			}

			continue
		}

		tp := utils.M(fields[fieldName])
		if tp == nil {
			tp = types.M{}
		}
		switch utils.S(tp["type"]) {
		case "Date":
			if v := utils.M(object[fieldName]); v != nil && utils.S(v["iso"]) != "" {
				valuesArray = append(valuesArray, v["iso"])
			} else {
				valuesArray = append(valuesArray, nil)
			}
		case "Pointer":
			if v := utils.M(object[fieldName]); v != nil && utils.S(v["objectId"]) != "" {
				valuesArray = append(valuesArray, v["objectId"])
			} else {
				valuesArray = append(valuesArray, "")
			}
		case "Array":
			b, err := json.Marshal(object[fieldName])
			if err != nil {
				return err
			}
			if fieldName == "_rperm" || fieldName == "_wperm" {
				// '[' => '{'
				if b[0] == 91 {
					b[0] = 123
				}
				// ']' => '}'
				if len(b) > 0 && b[len(b)-1] == 93 {
					b[len(b)-1] = 125
				}
			}
			valuesArray = append(valuesArray, b)
		case "Object":
			b, err := json.Marshal(object[fieldName])
			if err != nil {
				return err
			}
			valuesArray = append(valuesArray, b)
		case "String", "Number", "Boolean":
			valuesArray = append(valuesArray, object[fieldName])
		case "File":
			if v := utils.M(object[fieldName]); v != nil && utils.S(v["name"]) != "" {
				valuesArray = append(valuesArray, v["name"])
			} else {
				valuesArray = append(valuesArray, "")
			}
		case "GeoPoint":
			geoPoints[fieldName] = object[fieldName]
			columnsArray = columnsArray[:len(columnsArray)-1]
		default:
			return errs.E(errs.OtherCause, "Type "+utils.S(tp["type"])+" not supported yet")
		}

	}

	for key := range geoPoints {
		columnsArray = append(columnsArray, key)
	}

	initialValues := []string{}
	for index := range valuesArray {
		termination := ""
		fieldName := columnsArray[index]
		if fieldName == "_rperm" || fieldName == "_wperm" {
			termination = "::text[]"
		} else {
			fields := utils.M(schema["fields"])
			if fields == nil {
				fields = types.M{}
			}
			tp := utils.M(fields[fieldName])
			if tp == nil {
				tp = types.M{}
			}
			if utils.S(tp["type"]) == "Array" {
				termination = "::jsonb"
			}
		}
		initialValues = append(initialValues, fmt.Sprintf(`$%d%s`, index+1, termination))
	}

	geoPointsInjects := []string{}
	for _, v := range geoPoints {
		value := utils.M(v)
		if value == nil {
			value = types.M{}
		}
		valuesArray = append(valuesArray, value["longitude"], value["latitude"])
		l := len(valuesArray)
		geoPointsInjects = append(geoPointsInjects, fmt.Sprintf(`POINT($%d, $%d)`, l-1, l))
	}

	columnsPatternArray := []string{}
	for _, key := range columnsArray {
		columnsPatternArray = append(columnsPatternArray, fmt.Sprintf(`"%s"`, key))
	}
	columnsPattern := strings.Join(columnsPatternArray, ",")

	initialValues = append(initialValues, geoPointsInjects...)
	valuesPattern := strings.Join(initialValues, ",")

	qs := fmt.Sprintf(`INSERT INTO "%s" (%s) VALUES (%s)`, className, columnsPattern, valuesPattern)
	_, err = p.db.Exec(qs, valuesArray...)
	if err != nil {
		if e, ok := err.(*pq.Error); ok {
			if e.Code == postgresUniqueIndexViolationError {
				return errs.E(errs.DuplicateValue, "A duplicate value for a field with unique values was provided")
			}
		}
		return err
	}

	return nil
}

// GetAllClasses ...
func (p *PostgresAdapter) GetAllClasses() ([]types.M, error) {
	err := p.ensureSchemaCollectionExists()
	if err != nil {
		return nil, err
	}
	qs := `SELECT "className","schema" FROM "_SCHEMA"`
	rows, err := p.db.Query(qs)
	if err != nil {
		return nil, err
	}

	schemas := []types.M{}

	for rows.Next() {
		var clsName string
		var sch types.M
		var v []byte
		err := rows.Scan(&clsName, &v)
		if err != nil {
			return nil, err
		}
		err = json.Unmarshal(v, &sch)
		if err != nil {
			return nil, err
		}
		sch["className"] = clsName
		schemas = append(schemas, toParseSchema(sch))
	}

	return schemas, nil
}

// GetClass ...
func (p *PostgresAdapter) GetClass(className string) (types.M, error) {
	err := p.ensureSchemaCollectionExists()
	if err != nil {
		return nil, err
	}
	qs := `SELECT "schema" FROM "_SCHEMA" WHERE "className"=$1`
	rows, err := p.db.Query(qs, className)
	if err != nil {
		return nil, err
	}

	schema := types.M{}
	if rows.Next() {
		var v []byte
		err = rows.Scan(&v)
		if err != nil {
			return nil, err
		}
		err = json.Unmarshal(v, &schema)
		if err != nil {
			return nil, err
		}
	} else {
		return schema, nil
	}

	return toParseSchema(schema), nil
}

// DeleteObjectsByQuery 删除符合条件的所有对象
func (p *PostgresAdapter) DeleteObjectsByQuery(className string, schema, query types.M) error {
	where, err := buildWhereClause(schema, query, 1)
	if err != nil {
		return err
	}

	if len(query) == 0 {
		where.pattern = "TRUE"
	}

	qs := fmt.Sprintf(`WITH deleted AS (DELETE FROM "%s" WHERE %s RETURNING *) SELECT count(*) FROM deleted`, className, where.pattern)
	row := p.db.QueryRow(qs, where.values...)
	var count int
	err = row.Scan(&count)
	if err != nil {
		if e, ok := err.(*pq.Error); ok {
			// 表不存在返回空
			if e.Code == postgresRelationDoesNotExistError {
				return errs.E(errs.ObjectNotFound, "Object not found.")
			}
		}
		return err
	}

	if count == 0 {
		return errs.E(errs.ObjectNotFound, "Object not found.")
	}

	return nil
}

// Find ...
func (p *PostgresAdapter) Find(className string, schema, query, options types.M) ([]types.M, error) {
	if schema == nil {
		schema = types.M{}
	}
	if options == nil {
		options = types.M{}
	}

	var hasLimit bool
	var hasSkip bool
	if _, ok := options["limit"]; ok {
		hasLimit = true
	}
	if _, ok := options["skip"]; ok {
		hasSkip = true
	}

	values := types.S{}
	where, err := buildWhereClause(schema, query, 1)
	if err != nil {
		return nil, err
	}
	values = append(values, where.values...)

	var wherePattern string
	var limitPattern string
	var skipPattern string
	if where.pattern != "" {
		wherePattern = `WHERE ` + where.pattern
	}
	if hasLimit {
		limitPattern = fmt.Sprintf(`LIMIT $%d`, len(values)+1)
		values = append(values, options["limit"])
	}
	if hasSkip {
		skipPattern = fmt.Sprintf(`OFFSET $%d`, len(values)+1)
		values = append(values, options["skip"])
	}

	var sortPattern string
	if _, ok := options["sort"]; ok {
		if keys, ok := options["sort"].([]string); ok {
			postgresSort := []string{}
			for _, key := range keys {
				var postgresKey string
				if strings.HasPrefix(key, "-") {
					key = key[1:]
					postgresKey = fmt.Sprintf(`"%s" DESC`, key)
				} else {
					postgresKey = fmt.Sprintf(`"%s" ASC`, key)
				}
				postgresSort = append(postgresSort, postgresKey)
			}
			sorting := strings.Join(postgresSort, ",")
			if len(postgresSort) > 0 {
				sortPattern = fmt.Sprintf(`ORDER BY %s`, sorting)
			}
		}
	}
	if len(where.sorts) > 0 {
		sortPattern = fmt.Sprintf(`ORDER BY %s`, strings.Join(where.sorts, ","))
	}

	columns := "*"
	if _, ok := options["keys"]; ok {
		if keys, ok := options["keys"].([]string); ok {
			postgresKeys := []string{}
			for _, key := range keys {
				if key != "" {
					postgresKeys = append(postgresKeys, fmt.Sprintf(`"%s"`, key))
				}
			}
			if len(postgresKeys) > 0 {
				columns = strings.Join(postgresKeys, ",")
			}
		}
	}

	qs := fmt.Sprintf(`SELECT %s FROM "%s" %s %s %s %s`, columns, className, wherePattern, sortPattern, limitPattern, skipPattern)
	rows, err := p.db.Query(qs, values...)
	if err != nil {
		if e, ok := err.(*pq.Error); ok {
			// 表不存在返回空
			if e.Code == postgresRelationDoesNotExistError {
				return []types.M{}, nil
			}
		}
		return nil, err
	}

	fields := utils.M(schema["fields"])
	if fields == nil {
		fields = types.M{}
	}

	results := []types.M{}
	var resultColumns []string
	for rows.Next() {
		if resultColumns == nil {
			resultColumns, err = rows.Columns()
			if err != nil {
				return nil, err
			}
		}
		resultValues := []*interface{}{}
		values := types.S{}
		for i := 0; i < len(resultColumns); i++ {
			var v interface{}
			resultValues = append(resultValues, &v)
			values = append(values, &v)
		}
		err = rows.Scan(values...)
		if err != nil {
			return nil, err
		}
		object := types.M{}
		for i, field := range resultColumns {
			object[field] = *resultValues[i]
		}

		object, err = postgresObjectToParseObject(object, fields)
		if err != nil {
			return nil, err
		}

		results = append(results, object)
	}

	return results, nil
}

// Count ...
func (p *PostgresAdapter) Count(className string, schema, query types.M) (int, error) {
	where, err := buildWhereClause(schema, query, 1)
	if err != nil {
		return 0, err
	}

	wherePattern := ""
	if len(where.pattern) > 0 {
		wherePattern = `WHERE ` + where.pattern
	}

	qs := fmt.Sprintf(`SELECT count(*) FROM "%s" %s`, className, wherePattern)
	rows, err := p.db.Query(qs, where.values...)
	if err != nil {
		if e, ok := err.(*pq.Error); ok {
			if e.Code == postgresRelationDoesNotExistError {
				return 0, nil
			}
		}
		return 0, err
	}
	var count int
	if rows.Next() {
		err = rows.Scan(&count)
		if err != nil {
			return 0, nil
		}
	}

	return count, nil
}

// UpdateObjectsByQuery ...
func (p *PostgresAdapter) UpdateObjectsByQuery(className string, schema, query, update types.M) error {
	_, err := p.FindOneAndUpdate(className, schema, query, update)
	return err
}

// FindOneAndUpdate ...
func (p *PostgresAdapter) FindOneAndUpdate(className string, schema, query, update types.M) (types.M, error) {
	updatePatterns := []string{}
	values := types.S{}
	index := 1

	if schema == nil {
		schema = types.M{}
	}
	schema = toPostgresSchema(schema)

	fields := utils.M(schema["fields"])
	if fields == nil {
		fields = types.M{}
	}

	originalUpdate := utils.CopyMapM(update)
	update = handleDotFields(update)

	for fieldName, v := range update {
		re := regexp.MustCompile(`^_auth_data_([a-zA-Z0-9_]+)$`)
		authDataMatch := re.FindStringSubmatch(fieldName)
		if authDataMatch != nil && len(authDataMatch) == 2 {
			provider := authDataMatch[1]
			delete(update, fieldName)
			authData := utils.M(update["authData"])
			if authData == nil {
				authData = types.M{}
			}
			authData[provider] = v
			update["authData"] = authData
		}
	}

	for fieldName, fieldValue := range update {
		if fieldValue == nil {
			updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = NULL`, fieldName))
			continue
		}

		if fieldName == "authData" {
			generate := func(jsonb, key, value string) string {
				return fmt.Sprintf(`json_object_set_key(COALESCE(%s, '{}'::jsonb), %s, %s)::jsonb`, jsonb, key, value)
			}
			lastKey := fmt.Sprintf(`"%s"`, fieldName)
			authData := utils.M(fieldValue)
			if authData == nil {
				continue
			}
			for key, value := range authData {
				lastKey = generate(lastKey, fmt.Sprintf(`$%d::text`, index), fmt.Sprintf(`$%d::jsonb`, index+1))
				index = index + 2
				if value != nil {
					if v := utils.M(value); v != nil && utils.S(v["__op"]) == "Delete" {
						value = nil
					} else {
						b, err := json.Marshal(v)
						if err != nil {
							return nil, err
						}
						value = string(b)
					}
				}
				values = append(values, key, value)
			}
			updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = %s`, fieldName, lastKey))
			continue
		}

		if fieldName == "updatedAt" {
			updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = $%d`, fieldName, index))
			values = append(values, fieldValue)
			index = index + 1
			continue
		}

		switch fieldValue.(type) {
		case string, bool, float64, int:
			updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = $%d`, fieldName, index))
			values = append(values, fieldValue)
			index = index + 1
			continue
		case time.Time:
			updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = $%d`, fieldName, index))
			values = append(values, utils.TimetoString(fieldValue.(time.Time)))
			index = index + 1
			continue
		}

		if object := utils.M(fieldValue); object != nil {
			switch utils.S(object["__op"]) {
			case "Increment":
				updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = COALESCE("%s", 0) + $%d`, fieldName, fieldName, index))
				values = append(values, object["amount"])
				index = index + 1
				continue
			case "Add":
				updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = array_add(COALESCE("%s", '[]'::jsonb), $%d::jsonb)`, fieldName, fieldName, index))
				b, err := json.Marshal(object["objects"])
				if err != nil {
					return nil, err
				}
				values = append(values, string(b))
				index = index + 1
				continue
			case "Delete":
				updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = $%d`, fieldName, index))
				values = append(values, nil)
				index = index + 1
				continue
			case "Remove":
				updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = array_remove(COALESCE("%s", '[]'::jsonb), $%d::jsonb)`, fieldName, fieldName, index))
				b, err := json.Marshal(object["objects"])
				if err != nil {
					return nil, err
				}
				values = append(values, string(b))
				index = index + 1
				continue
			case "AddUnique":
				updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = array_add_unique(COALESCE("%s", '[]'::jsonb), $%d::jsonb)`, fieldName, fieldName, index))
				b, err := json.Marshal(object["objects"])
				if err != nil {
					return nil, err
				}
				values = append(values, string(b))
				index = index + 1
				continue
			}

			switch utils.S(object["__type"]) {
			case "Pointer":
				updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = $%d`, fieldName, index))
				values = append(values, object["objectId"])
				index = index + 1
				continue
			case "Date", "File":
				updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = $%d`, fieldName, index))
				values = append(values, toPostgresValue(object))
				index = index + 1
				continue
			case "GeoPoint":
				updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = POINT($%d, $%d)`, fieldName, index, index+1))
				values = append(values, object["longitude"], object["latitude"])
				index = index + 2
				continue
			case "Relation":
				continue
			}

			if tp := utils.M(fields[fieldName]); tp != nil && utils.S(tp["type"]) == "Object" {
				keysToDelete := []string{}
				for k, v := range originalUpdate {
					if o := utils.M(v); o != nil && utils.S(o["__op"]) == "Delete" {
						if keys := strings.Split(k, "."); len(keys) == 2 && keys[0] == fieldName {
							keysToDelete = append(keysToDelete, keys[1])
						}
					}
				}

				deletePatterns := ""
				for _, k := range keysToDelete {
					deletePatterns = deletePatterns + fmt.Sprintf(` - '%s'`, k)
				}

				updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = ( COALESCE("%s", '{}'::jsonb) %s || $%d::jsonb )`, fieldName, fieldName, deletePatterns, index))
				b, err := json.Marshal(object)
				if err != nil {
					return nil, err
				}
				values = append(values, string(b))
				index = index + 1
				continue
			}
		}

		if array := utils.A(fieldValue); array != nil {
			if tp := utils.M(fields[fieldName]); tp != nil && utils.S(tp["type"]) == "Array" {
				expectedType, err := parseTypeToPostgresType(tp)
				if err != nil {
					return nil, err
				}

				b, err := json.Marshal(fieldValue)
				if err != nil {
					return nil, err
				}

				if expectedType == "text[]" {
					// '[' => '{'
					if b[0] == 91 {
						b[0] = 123
					}
					// ']' => '}'
					if len(b) > 0 && b[len(b)-1] == 93 {
						b[len(b)-1] = 125
					}
					updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = $%d::text[]`, fieldName, index))
				} else {
					updatePatterns = append(updatePatterns, fmt.Sprintf(`"%s" = $%d::jsonb`, fieldName, index))
				}

				values = append(values, string(b))
				index = index + 1
				continue
			}
		}

		b, _ := json.Marshal(fieldValue)
		return nil, errs.E(errs.OperationForbidden, "Postgres doesn't support update "+string(b)+" yet")
	}

	where, err := buildWhereClause(schema, query, index)
	if err != nil {
		return nil, err
	}
	values = append(values, where.values...)

	// TODO 需要添加限制，只更新一条，UpdateObjectsByQuery 时更新多条
	qs := fmt.Sprintf(`UPDATE "%s" SET %s WHERE %s RETURNING *`, className, strings.Join(updatePatterns, ","), where.pattern)
	rows, err := p.db.Query(qs, values...)
	if err != nil {
		if e, ok := err.(*pq.Error); ok {
			// 表不存在返回空
			if e.Code == postgresRelationDoesNotExistError {
				return nil, errs.E(errs.ObjectNotFound, "Object not found.")
			}
		}
		return nil, err
	}
{'fraction_non_alphanumeric': 0.16384625006701095, 'fraction_numerical': 0.005253658798091528, 'mean_word_length': 3.3874558996471973, 'pattern_counts': {'":': 56, '<': 5, '<?xml version=': 0, '>': 25, 'https://': 0, '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': '966021', 'n_tokens_mistral': 22701, 'n_tokens_neox': 20083, 'n_words': 4894}