// Grants returns a list of SQL queries that constitute the privileges specified
// in the receiver for the target schema.
func (s Schema) Grants(target string) []string {
	const maxQueries = 2
	queries := make([]string, 0, maxQueries)

	if s.GetPrivilege(Create) {
		b := bytes.NewBufferString("GRANT CREATE ON SCHEMA ")
		fmt.Fprint(b, pq.QuoteIdentifier(target), " TO ", quoteRole(s.Role))

		if s.GetGrantOption(Create) {
			fmt.Fprint(b, " WITH GRANT OPTION")
		}

		queries = append(queries, b.String())
	}

	if s.GetPrivilege(Usage) {
		b := bytes.NewBufferString("GRANT USAGE ON SCHEMA ")
		fmt.Fprint(b, pq.QuoteIdentifier(target), " TO ", quoteRole(s.Role))

		if s.GetGrantOption(Usage) {
			fmt.Fprint(b, " WITH GRANT OPTION")
		}

		queries = append(queries, b.String())
	}

	return queries
}

// Revokes returns a list of SQL queries that remove the privileges specified
// in the receiver from the target schema.
func (s Schema) Revokes(target string) []string {
	const maxQueries = 2
	queries := make([]string, 0, maxQueries)

	if s.GetPrivilege(Create) {
		b := bytes.NewBufferString("REVOKE")
		if s.GetGrantOption(Create) {
			fmt.Fprint(b, " GRANT OPTION FOR")
		}

		fmt.Fprint(b, " CREATE ON SCHEMA ")
		fmt.Fprint(b, pq.QuoteIdentifier(target), " FROM ", quoteRole(s.Role))
		queries = append(queries, b.String())
	}

	if s.GetPrivilege(Usage) {
		b := bytes.NewBufferString("REVOKE")
		if s.GetGrantOption(Usage) {
			fmt.Fprint(b, " GRANT OPTION FOR")
		}

		fmt.Fprint(b, " USAGE ON SCHEMA ")
		fmt.Fprint(b, pq.QuoteIdentifier(target), " FROM ", quoteRole(s.Role))
		queries = append(queries, b.String())
	}

	return queries
}

func resourcePostgresqlDatabaseCreate(d *schema.ResourceData, meta interface{}) error {
	client := meta.(*Client)
	conn, err := client.Connect()
	if err != nil {
		return err
	}
	defer conn.Close()

	dbName := d.Get("name").(string)
	dbOwner := d.Get("owner").(string)
	connUsername := client.username

	var dbOwnerCfg string
	if dbOwner != "" {
		dbOwnerCfg = fmt.Sprintf("WITH OWNER=%s", pq.QuoteIdentifier(dbOwner))
	} else {
		dbOwnerCfg = ""
	}

	//needed in order to set the owner of the db if the connection user is not a superuser
	err = grantRoleMembership(conn, dbOwner, connUsername)
	if err != nil {
		return err
	}

	query := fmt.Sprintf("CREATE DATABASE %s %s", pq.QuoteIdentifier(dbName), dbOwnerCfg)
	_, err = conn.Query(query)
	if err != nil {
		return fmt.Errorf("Error creating postgresql database %s: %s", dbName, err)
	}

	d.SetId(dbName)

	return nil
}

func resourcePostgreSQLExtensionCreate(d *schema.ResourceData, meta interface{}) error {
	c := meta.(*Client)
	c.catalogLock.Lock()
	defer c.catalogLock.Unlock()

	conn, err := c.Connect()
	if err != nil {
		return err
	}
	defer conn.Close()

	extName := d.Get(extNameAttr).(string)

	b := bytes.NewBufferString("CREATE EXTENSION ")
	fmt.Fprint(b, pq.QuoteIdentifier(extName))

	if v, ok := d.GetOk(extSchemaAttr); ok {
		fmt.Fprint(b, " SCHEMA ", pq.QuoteIdentifier(v.(string)))
	}

	if v, ok := d.GetOk(extVersionAttr); ok {
		fmt.Fprint(b, " VERSION ", pq.QuoteIdentifier(v.(string)))
	}

	query := b.String()
	_, err = conn.Query(query)
	if err != nil {
		return errwrap.Wrapf("Error creating extension: {{err}}", err)
	}

	d.SetId(extName)

	return resourcePostgreSQLExtensionReadImpl(d, meta)
}

func (config *Config) quotedMigrationTable() string {
	if !strings.Contains(config.MigrationTable, ".") {
		return pq.QuoteIdentifier(config.MigrationTable)
	}

	tokens := strings.SplitN(config.MigrationTable, ".", 2)
	return pq.QuoteIdentifier(tokens[0]) + "." + pq.QuoteIdentifier(tokens[1])
}

// Create a new key/value
func NewKeyValue(host *Host, name string) (*KeyValue, error) {
	kv := &KeyValue{host, name}
	query := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (attr hstore)", pq.QuoteIdentifier(kvPrefix+kv.table))
	if _, err := kv.host.db.Exec(query); err != nil {
		if !strings.HasSuffix(err.Error(), "already exists") {
			return nil, err
		}
	}
	if Verbose {
		log.Println("Created HSTORE table " + pq.QuoteIdentifier(kvPrefix+kv.table) + " in database " + host.dbname)
	}
	return kv, nil
}

func SchemaInit(dbh *sql.DB, schema string) error {
	if schema == "" {
		schema = "request_dump"
	}
	if strings.Index(schema, " ") >= 0 {
		return fmt.Errorf("schemas containing a space are not supported")
	}

	// initialize schema where request data will be stored
	exists, err := SchemaExists(dbh, schema)
	if err != nil {
		return err
	}
	if exists == false {
		log.Printf("pg.SchemaInit: creating schema [%s]\n", schema)
		_, err := dbh.Exec(fmt.Sprintf("CREATE SCHEMA %s", pq.QuoteIdentifier(schema)))
		if err != nil {
			return fmt.Errorf("pg.SchemaInit: %s", err)
		}
	}

	table := "raw_requests"
	exists, err = TableExistsInSchema(dbh, table, schema)
	if err != nil {
		return err
	}
	if exists == false {
		log.Printf("pg.SchemaInit: creating table [%s.%s]\n", schema, table)
		ddls := []string{
			fmt.Sprintf(`
				CREATE TABLE %s.%s (
					request_id bigserial primary key,
					head       text,
					data       text,
					"when"     timestamptz,
					batch_id   bigint
				)
			`, pq.QuoteIdentifier(schema), table),
			fmt.Sprintf("CREATE INDEX raw_requests_batch_id_idx ON %s.%s (batch_id)",
				schema, table),
		}
		for _, ddl := range ddls {
			_, err := dbh.Exec(ddl)
			if err != nil {
				return fmt.Errorf("pg.SchemaInit: %s", err)
			}
		}
	}

	return nil
}

func resourcePostgresqlRoleUpdate(d *schema.ResourceData, meta interface{}) error {
	client := meta.(*Client)
	conn, err := client.Connect()
	if err != nil {
		return err
	}
	defer conn.Close()

	d.Partial(true)

	roleName := d.Get("name").(string)

	if d.HasChange("login") {
		loginAttr := getLoginStr(d.Get("login").(bool))
		query := fmt.Sprintf("ALTER ROLE %s %s", pq.QuoteIdentifier(roleName), pq.QuoteIdentifier(loginAttr))
		_, err := conn.Query(query)
		if err != nil {
			return fmt.Errorf("Error updating login attribute for role: %s", err)
		}

		d.SetPartial("login")
	}

	password := d.Get("password").(string)
	if d.HasChange("password") {
		encryptedCfg := getEncryptedStr(d.Get("encrypted").(bool))

		query := fmt.Sprintf("ALTER ROLE %s %s PASSWORD '%s'", pq.QuoteIdentifier(roleName), encryptedCfg, password)
		_, err := conn.Query(query)
		if err != nil {
			return fmt.Errorf("Error updating password attribute for role: %s", err)
		}

		d.SetPartial("password")
	}

	if d.HasChange("encrypted") {
		encryptedCfg := getEncryptedStr(d.Get("encrypted").(bool))

		query := fmt.Sprintf("ALTER ROLE %s %s PASSWORD '%s'", pq.QuoteIdentifier(roleName), encryptedCfg, password)
		_, err := conn.Query(query)
		if err != nil {
			return fmt.Errorf("Error updating encrypted attribute for role: %s", err)
		}

		d.SetPartial("encrypted")
	}

	d.Partial(false)
	return resourcePostgresqlRoleRead(d, meta)
}

func grantRoleMembership(conn *sql.DB, dbOwner string, connUsername string) error {
	if dbOwner != "" && dbOwner != connUsername {
		query := fmt.Sprintf("GRANT %s TO %s", pq.QuoteIdentifier(dbOwner), pq.QuoteIdentifier(connUsername))
		_, err := conn.Query(query)
		if err != nil {
			//is already member or role
			if strings.Contains(err.Error(), "duplicate key value violates unique constraint") {
				return nil
			}
			return fmt.Errorf("Error granting membership: %s", err)
		}
	}
	return nil
}

func resourcePostgresqlDatabaseDelete(d *schema.ResourceData, meta interface{}) error {
	client := meta.(*Client)
	conn, err := client.Connect()
	if err != nil {
		return err
	}
	defer conn.Close()

	dbName := d.Get("name").(string)
	connUsername := client.username
	dbOwner := d.Get("owner").(string)
	//needed in order to set the owner of the db if the connection user is not a superuser
	err = grantRoleMembership(conn, dbOwner, connUsername)
	if err != nil {
		return err
	}

	query := fmt.Sprintf("DROP DATABASE %s", pq.QuoteIdentifier(dbName))
	_, err = conn.Query(query)
	if err != nil {
		return err
	}

	d.SetId("")

	return nil
}

func createSchemaUnlessExists(c *Config, db *sql.DB) error {
	// If there's no schema name in the config, we don't need to create the schema.
	if !strings.Contains(c.MigrationTable, ".") {
		return nil
	}

	var exists bool

	schema := strings.SplitN(c.MigrationTable, ".", 2)[0]
	err := db.QueryRow(
		`SELECT EXISTS(SELECT 1 FROM pg_catalog.pg_namespace WHERE nspname = $1)`,
		schema,
	).Scan(&exists)

	if err != nil {
		return err
	}

	if exists {
		return nil
	}

	_, err = db.Exec(fmt.Sprintf(
		"CREATE SCHEMA %s;",
		pq.QuoteIdentifier(schema),
	))
	return err
}

func (r *userRepo) GetRemoteIdentities(tx repo.Transaction, userID string) ([]user.RemoteIdentity, error) {
	ex := r.executor(tx)
	if userID == "" {
		return nil, user.ErrorInvalidID
	}

	qt := pq.QuoteIdentifier(remoteIdentityMappingTableName)
	rims, err := ex.Select(&remoteIdentityMappingModel{},
		fmt.Sprintf("select * from %s where user_id = $1", qt), userID)

	if err != nil {
		if err != sql.ErrNoRows {
			return nil, err
		}
		return nil, err
	}
	if len(rims) == 0 {
		return nil, nil
	}

	var ris []user.RemoteIdentity
	for _, m := range rims {
		rim, ok := m.(*remoteIdentityMappingModel)
		if !ok {
			return nil, errors.New("unrecognized model")
		}

		ris = append(ris, user.RemoteIdentity{
			ID:          rim.RemoteID,
			ConnectorID: rim.ConnectorID,
		})
	}

	return ris, nil
}

func (b *backend) secretCredsRevoke(
	req *logical.Request, d *framework.FieldData) (*logical.Response, error) {
	// Get the username from the internal data
	usernameRaw, ok := req.Secret.InternalData["username"]
	if !ok {
		return nil, fmt.Errorf("secret is missing username internal data")
	}
	username, ok := usernameRaw.(string)

	// Get our connection
	db, err := b.DB(req.Storage)
	if err != nil {
		return nil, err
	}

	// Drop this user
	stmt, err := db.Prepare(fmt.Sprintf(
		"DROP ROLE IF EXISTS %s;", pq.QuoteIdentifier(username)))
	if err != nil {
		return nil, err
	}
	defer stmt.Close()
	if _, err := stmt.Exec(); err != nil {
		return nil, err
	}

	return nil, nil
}

func resourcePostgreSQLSchemaDelete(d *schema.ResourceData, meta interface{}) error {
	c := meta.(*Client)
	c.catalogLock.Lock()
	defer c.catalogLock.Unlock()

	conn, err := c.Connect()
	if err != nil {
		return err
	}
	defer conn.Close()

	txn, err := conn.Begin()
	if err != nil {
		return err
	}
	defer txn.Rollback()

	schemaName := d.Get(schemaNameAttr).(string)

	// NOTE([email protected]): Deliberately not performing a cascading drop.
	query := fmt.Sprintf("DROP SCHEMA %s", pq.QuoteIdentifier(schemaName))
	_, err = txn.Query(query)
	if err != nil {
		return errwrap.Wrapf("Error deleting schema: {{err}}", err)
	}

	if err := txn.Commit(); err != nil {
		return errwrap.Wrapf("Error committing schema: {{err}}", err)
	}

	d.SetId("")

	return nil
}

func resourcePostgreSQLDatabaseDelete(d *schema.ResourceData, meta interface{}) error {
	c := meta.(*Client)
	conn, err := c.Connect()
	if err != nil {
		return errwrap.Wrapf("Error connecting to PostgreSQL: {{err}}", err)
	}
	defer conn.Close()

	dbName := d.Get(dbNameAttr).(string)

	if isTemplate := d.Get(dbIsTemplateAttr).(bool); isTemplate {
		// Template databases must have this attribute cleared before
		// they can be dropped.
		if err := doSetDBIsTemplate(conn, dbName, false); err != nil {
			return errwrap.Wrapf("Error updating database IS_TEMPLATE during DROP DATABASE: {{err}}", err)
		}
	}

	if err := setDBIsTemplate(conn, d); err != nil {
		return err
	}

	query := fmt.Sprintf("DROP DATABASE %s", pq.QuoteIdentifier(dbName))
	_, err = conn.Query(query)
	if err != nil {
		return errwrap.Wrapf("Error dropping database: {{err}}", err)
	}

	d.SetId("")

	return nil
}

func (r *ConnectorConfigRepo) Set(cfgs []connector.ConnectorConfig) error {
	insert := make([]interface{}, len(cfgs))
	for i, cfg := range cfgs {
		m, err := newConnectorConfigModel(cfg)
		if err != nil {
			return err
		}

		insert[i] = m
	}

	tx, err := r.dbMap.Begin()
	if err != nil {
		return err
	}

	qt := pq.QuoteIdentifier(connectorConfigTableName)
	q := fmt.Sprintf("DELETE FROM %s", qt)
	if _, err = r.dbMap.Exec(q); err != nil {
		return err
	}

	if err = r.dbMap.Insert(insert...); err != nil {
		return fmt.Errorf("DB insert failed %#v: %v", insert, err)
	}

	return tx.Commit()
}

// quoteRole is a small helper function that handles the quoting of a role name,
// or PUBLIC, if no role is specified.
func quoteRole(role string) string {
	if role == "" {
		return "PUBLIC"
	}

	return pq.QuoteIdentifier(role)
}

func (r *SessionKeyRepo) Pop(key string) (string, error) {
	m, err := r.dbMap.Get(sessionKeyModel{}, key)
	if err != nil {
		return "", err
	}

	skm, ok := m.(*sessionKeyModel)
	if !ok {
		return "", errors.New("unrecognized model")
	}

	if skm.Stale || skm.ExpiresAt < r.clock.Now().Unix() {
		return "", errors.New("invalid session key")
	}

	qt := pq.QuoteIdentifier(sessionKeyTableName)
	q := fmt.Sprintf("UPDATE %s SET stale=$1 WHERE key=$2 AND stale=$3", qt)
	res, err := r.dbMap.Exec(q, true, key, false)
	if err != nil {
		return "", err
	}

	if n, err := res.RowsAffected(); n != 1 {
		if err != nil {
			log.Errorf("Failed determining rows affected by UPDATE session_key query: %v", err)
		}
		return "", fmt.Errorf("failed to pop entity")
	}

	return skm.SessionID, nil
}

// newPostgreSQLBackend constructs a PostgreSQL backend using the given
// API client, server address, credentials, and database.
func newPostgreSQLBackend(conf map[string]string, logger *log.Logger) (Backend, error) {
	// Get the PostgreSQL credentials to perform read/write operations.
	connURL, ok := conf["connection_url"]
	if !ok || connURL == "" {
		return nil, fmt.Errorf("missing connection_url")
	}

	unquoted_table, ok := conf["table"]
	if !ok {
		unquoted_table = "vault_kv_store"
	}
	quoted_table := pq.QuoteIdentifier(unquoted_table)

	// Create PostgreSQL handle for the database.
	db, err := sql.Open("postgres", connURL)
	if err != nil {
		return nil, fmt.Errorf("failed to connect to postgres: %v", err)
	}

	// Determine if we should use an upsert function (versions < 9.5)
	var upsert_required bool
	upsert_required_query := "SELECT string_to_array(setting, '.')::int[] < '{9,5}' FROM pg_settings WHERE name = 'server_version'"
	if err := db.QueryRow(upsert_required_query).Scan(&upsert_required); err != nil {
		return nil, fmt.Errorf("failed to check for native upsert: %v", err)
	}

	// Setup our put strategy based on the presence or absence of a native
	// upsert.
	var put_statement string
	if upsert_required {
		put_statement = "SELECT vault_kv_put($1, $2, $3, $4)"
	} else {
		put_statement = "INSERT INTO " + quoted_table + " VALUES($1, $2, $3, $4)" +
			" ON CONFLICT (path, key) DO " +
			" UPDATE SET (parent_path, path, key, value) = ($1, $2, $3, $4)"
	}

	// Setup the backend.
	m := &PostgreSQLBackend{
		table:      quoted_table,
		client:     db,
		statements: make(map[string]*sql.Stmt),
		logger:     logger,
	}

	// Prepare all the statements required
	statements := map[string]string{
		"put":    put_statement,
		"get":    "SELECT value FROM " + quoted_table + " WHERE path = $1 AND key = $2",
		"delete": "DELETE FROM " + quoted_table + " WHERE path = $1 AND key = $2",
		"list": "SELECT key FROM " + quoted_table + " WHERE path = $1" +
			"UNION SELECT substr(path, length($1)+1) FROM " + quoted_table + "WHERE parent_path = $1",
	}
	for name, query := range statements {
		if err := m.prepare(name, query); err != nil {
			return nil, err
		}
	}
	return m, nil
}

// TruncateTable does horrible things which is why it's only allowed on test databases
func (db *TestDB) TruncateTable(tableName string) {
	query := fmt.Sprintf("TRUNCATE TABLE %s CASCADE", pq.QuoteIdentifier(tableName))
	_, err := db.Exec(query)
	if err != nil {
		log.Fatal(err)
	}
}