// OpenFromConnPool takes the existing *pgx.ConnPool pool and returns a *sql.DB
// with pool as the backend. This enables full control over the connection
// process and configuration while maintaining compatibility with the
// database/sql interface. In addition, by calling Driver() on the returned
// *sql.DB and typecasting to *stdlib.Driver a reference to the pgx.ConnPool can
// be reaquired later. This allows fast paths targeting pgx to be used while
// still maintaining compatibility with other databases and drivers.
//
// pool connection size must be at least 2.
func OpenFromConnPool(pool *pgx.ConnPool) (*sql.DB, error) {
	d := &Driver{Pool: pool}
	name := fmt.Sprintf("pgx-%d", openFromConnPoolCount)
	openFromConnPoolCount++
	sql.Register(name, d)
	db, err := sql.Open(name, "")
	if err != nil {
		return nil, err
	}

	// Presumably OpenFromConnPool is being used because the user wants to use
	// database/sql most of the time, but fast path with pgx some of the time.
	// Allow database/sql to use all the connections, but release 2 idle ones.
	// Don't have database/sql immediately release all idle connections because
	// that would mean that prepared statements would be lost (which kills
	// performance if the prepared statements constantly have to be reprepared)
	stat := pool.Stat()

	if stat.MaxConnections <= 2 {
		return nil, errors.New("pool connection size must be at least 2")
	}
	db.SetMaxIdleConns(stat.MaxConnections - 2)
	db.SetMaxOpenConns(stat.MaxConnections)

	return db, nil
}

func FindModel(model models.Model, db *pgx.ConnPool, tx *pgx.Tx, where string, fields ...string) error {
	fieldNames, fieldValues := model.Fields(fields...)

	query := SqlSelect(model.TableName(), fieldNames)

	where = fmt.Sprintf(" WHERE %s = ? %s", model.PrimaryName(), where)

	query = FormateToPQuery(query + where)
	args := []interface{}{model.PrimaryValue()}

	var err error

	if tx != nil {
		err = tx.QueryRow(query, args...).Scan(fieldValues...)
	} else {
		err = db.QueryRow(query, args...).Scan(fieldValues...)
	}

	if err == pgx.ErrNoRows {

		return models.ErrNotFound
	}

	if err != nil {

		return err
	}

	return nil
}

func UpdateFeedWithFetchSuccess(db *pgx.ConnPool, feedID int32, update *ParsedFeed, etag String, fetchTime time.Time) error {
	_, err := db.Prepare("updateFeedWithFetchSuccess", updateFeedWithFetchSuccessSQL)
	if err != nil {
		return err
	}

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

	_, err = tx.Exec("updateFeedWithFetchSuccess",
		update.Name,
		fetchTime,
		&etag,
		feedID)
	if err != nil {
		return err
	}

	if len(update.Items) > 0 {
		insertSQL, insertArgs := buildNewItemsSQL(feedID, update.Items)
		_, err = tx.Exec(insertSQL, insertArgs...)
		if err != nil {
			return err
		}
	}

	return tx.Commit()
}

// Empty all data in the entire database
func empty(pool *pgx.ConnPool) error {
	tables := []string{"feeds", "items", "password_resets", "sessions", "subscriptions", "unread_items", "users"}
	for _, table := range tables {
		_, err := pool.Exec(fmt.Sprintf("delete from %s", table))
		if err != nil {
			return err
		}
	}
	return nil
}

func acquireAllConnections(t *testing.T, pool *pgx.ConnPool, maxConnections int) []*pgx.Conn {
	connections := make([]*pgx.Conn, maxConnections)
	for i := 0; i < maxConnections; i++ {
		var err error
		if connections[i], err = pool.Acquire(); err != nil {
			t.Fatalf("Unable to acquire connection: %v", err)
		}
	}
	return connections
}

func setupStressDB(t *testing.T, pool *pgx.ConnPool) {
	_, err := pool.Exec(`
		drop table if exists widgets;
		create table widgets(
			id serial primary key,
			name varchar not null,
			description text,
			creation_time timestamptz
		);
`)
	if err != nil {
		t.Fatal(err)
	}
}

func txInsertRollback(pool *pgx.ConnPool, actionNum int) error {
	tx, err := pool.Begin()
	if err != nil {
		return err
	}

	sql := `
		insert into widgets(name, description, creation_time)
		values($1, $2, $3)`

	_, err = tx.Exec(sql, fake.ProductName(), fake.Sentences(), time.Now())
	if err != nil {
		return err
	}

	return tx.Rollback()
}

func listenAndPoolUnlistens(pool *pgx.ConnPool, actionNum int) error {
	conn, err := pool.Acquire()
	if err != nil {
		return err
	}
	defer pool.Release(conn)

	err = conn.Listen("stress")
	if err != nil {
		return err
	}

	_, err = conn.WaitForNotification(100 * time.Millisecond)
	if err == pgx.ErrNotificationTimeout {
		return nil
	}
	return err
}

func updateOrCreateModel(model models.Model, db *pgx.ConnPool, tx *pgx.Tx, isNew, isRemove bool, where string, fields ...string) error {
	if isNew {
		model.BeforeCreate()
	}

	if isRemove {
		model.BeforeDelete()
	}

	model.BeforeSave()

	fieldNames, fieldValues := model.Fields(fields...)

	var query string = SqlUpdate(model.TableName(), fieldNames)

	if isNew {
		query = SqlInsert(model.TableName(), fieldNames)
	}

	where = fmt.Sprintf(" WHERE %[1]s = ? %[2]s RETURNING %[1]s", model.PrimaryName(), where)
	if isNew {
		where = fmt.Sprintf(" RETURNING %s", model.PrimaryName())
	} else {
		fieldValues = append(fieldValues, model.PrimaryValue())
	}

	query = FormateToPQuery(query + where)

	var err error

	if tx != nil {
		err = tx.QueryRow(query, fieldValues...).Scan(model.Maps()[model.PrimaryName()])

	} else {
		err = db.QueryRow(query, fieldValues...).Scan(model.Maps()[model.PrimaryName()])
	}

	if err != nil {

		return err
	}

	return nil
}

func DeleteSubscription(db *pgx.ConnPool, userID, feedID int32) error {
	if _, err := db.Prepare("deleteSubscription", deleteSubscriptionSQL); err != nil {
		return err
	}
	if _, err := db.Prepare("deleteFeedIfOrphaned", deleteFeedIfOrphanedSQL); err != nil {
		return err
	}

	tx, err := db.BeginIso(pgx.Serializable)
	if err != nil {
		return err
	}
	defer tx.Rollback()

	_, err = tx.Exec("deleteSubscription", userID, feedID)
	if err != nil {
		return err
	}

	_, err = tx.Exec("deleteFeedIfOrphaned", feedID)
	if err != nil {
		return err
	}

	return tx.Commit()
}

func txMultipleQueries(pool *pgx.ConnPool, actionNum int) error {
	tx, err := pool.Begin()
	if err != nil {
		return err
	}
	defer tx.Rollback()

	errExpectedTxDeath := errors.New("Expected tx death")

	actions := []struct {
		name string
		fn   func() error
	}{
		{"insertUnprepared", func() error { return insertUnprepared(tx, actionNum) }},
		{"queryRowWithoutParams", func() error { return queryRowWithoutParams(tx, actionNum) }},
		{"query", func() error { return query(tx, actionNum) }},
		{"queryCloseEarly", func() error { return queryCloseEarly(tx, actionNum) }},
		{"queryErrorWhileReturningRows", func() error {
			err := queryErrorWhileReturningRows(tx, actionNum)
			if err != nil {
				return err
			}
			return errExpectedTxDeath
		}},
	}

	for i := 0; i < 20; i++ {
		action := actions[rand.Intn(len(actions))]
		err := action.fn()
		if err == errExpectedTxDeath {
			return nil
		} else if err != nil {
			return err
		}
	}

	return tx.Commit()
}

func poolPrepareUseAndDeallocate(pool *pgx.ConnPool, actionNum int) error {
	psName := fmt.Sprintf("poolPreparedStatement%d", actionNum)

	_, err := pool.Prepare(psName, "select $1::text")
	if err != nil {
		return err
	}

	var s string
	err = pool.QueryRow(psName, "hello").Scan(&s)
	if err != nil {
		return err
	}

	if s != "hello" {
		return fmt.Errorf("Prepared statement did not return expected value: %v", s)
	}

	return pool.Deallocate(psName)
}

func acquireWithTimeTaken(pool *pgx.ConnPool) (*pgx.Conn, time.Duration, error) {
	startTime := time.Now()
	c, err := pool.Acquire()
	return c, time.Now().Sub(startTime), err
}

func releaseAllConnections(pool *pgx.ConnPool, connections []*pgx.Conn) {
	for _, c := range connections {
		pool.Release(c)
	}
}

func unlistenAndRelease(pool *pgx.ConnPool, conn *pgx.Conn, channel string) {
	if err := conn.Unlisten(channel); err != nil {
		conn.Close()
	}
	pool.Release(conn)
}

func truncateAndClose(pool *pgx.ConnPool) {
	if _, err := pool.Exec("TRUNCATE TABLE que_jobs"); err != nil {
		panic(err)
	}
	pool.Close()
}

func notify(pool *pgx.ConnPool, actionNum int) error {
	_, err := pool.Exec("notify stress")
	return err
}