package postgres

import (
	"context"
	"database/sql"
	"time"

	"code.justin.tv/eventbus/controlplane/internal/db"
	"github.com/jmoiron/sqlx"
	"github.com/pkg/errors"
)

const (
	EventStreamsTableName = "event_streams"

	eventStreamSelectBase = `
SELECT 
	event_streams.id,
	event_streams.environment,
	event_streams.event_type_id,
	event_streams.sns_topic_arn "sns_details.sns_topic_arn",
	event_types.name "event_type.name",
	event_types.schema "event_type.schema",
	event_types.repo_filepath "event_type.repo_filepath",
	event_types.deprecated "event_type.deprecated",
	event_types.ldap_group "event_type.ldap_group"
`

	// event streams get initialized with an empty SNS arn field (DONT SET IT TO NULL)
	eventStreamCreateQuery = `
INSERT INTO event_streams (environment, sns_topic_arn, event_type_id)
VALUES (:environment, '', :event_type_id) RETURNING id`

	eventStreamAllQuery = eventStreamSelectBase + `
FROM 
	event_streams
INNER JOIN event_types
	ON event_types.id = event_streams.event_type_id
ORDER BY event_types.name, event_streams.environment;
`

	eventStreamByIDQuery = eventStreamSelectBase + `
FROM 
	event_streams, 
	event_types
WHERE
	event_types.id = event_streams.event_type_id
		AND
	event_streams.id = $1
`

	eventStreamWithEnvQuery = eventStreamSelectBase + `
FROM 
	event_streams
INNER JOIN event_types
	ON event_types.id = event_streams.event_type_id
WHERE
	environment = $1
ORDER BY event_types.name;
`
	eventStreamWithNameQuery = eventStreamSelectBase + `
FROM 
	event_streams
INNER JOIN event_types
	ON event_types.id = event_streams.event_type_id
WHERE
	name = $1
ORDER BY event_types.name;
`

	eventStreamWithNameAndEnvQuery = eventStreamSelectBase + `
FROM 
	event_streams,
	event_types
WHERE
	event_types.id = event_streams.event_type_id
		AND
	environment = $1
		AND
	event_types.name = $2;
`

	eventStreamByEventTypeIDQuery = eventStreamSelectBase + `
FROM 
	event_streams
INNER JOIN 
	event_types ON event_types.id = event_streams.event_type_id
WHERE
	event_types.id = $1
`

	eventStreamUpdateInfraQuery = `
UPDATE event_streams
SET sns_topic_arn = $2
WHERE id = $1
`

	eventStreamDeleteByEventTypeIDQuery = `
DELETE from event_streams
WHERE event_type_id = $1
`
)

func (pg *PostgresDB) EventStreamCreate(ctx context.Context, eventStream *db.EventStream) (int, error) {
	var tx *sqlx.Tx
	var err error
	var res int
	if tx, err = pg.newTx(); err != nil {
		return -1, err
	}
	if res, err = pg.eventStreamCreateTx(ctx, tx, eventStream); err != nil {
		return -1, err
	}
	if err = pg.doneTx(tx); err != nil {
		return -1, err
	}
	return res, nil
}

func (pg *PostgresDB) eventStreamCreateTx(ctx context.Context, tx *sqlx.Tx, eventStream *db.EventStream) (int, error) {
	var id int
	namedQuery, err := tx.PrepareNamedContext(ctx, eventStreamCreateQuery)
	if err != nil {
		return -1, err
	}

	err = namedQuery.QueryRowxContext(ctx, eventStream).Scan(&id)
	return id, err
}

func (pg *PostgresDB) EventStreamUpdate(ctx context.Context, id int, eventStreamEditable *db.EventStreamEditable) (int, error) {
	return -1, db.ErrNotSupported
}

func (pg *PostgresDB) EventStreamUpdateInfra(ctx context.Context, lease db.AWSLease, id int, eventStreamInfraUpdate *db.EventStreamInfraUpdate) (int, error) {
	if lease == nil {
		return -1, errors.New("nil lease object passed into EventStreamsTable.Update")
	}

	if lease.Expired() {
		return -1, db.ErrLeaseExpired
	}

	result, err := pg.writer.ExecContext(ctx, eventStreamUpdateInfraQuery, id, eventStreamInfraUpdate.SNSTopicARN)
	if err != nil {
		return -1, err
	}
	i, err := result.RowsAffected()
	if err != nil {
		return -1, err
	} else if i == 0 {
		return -1, db.ErrResourceNotFound
	}
	return id, nil
}

func (pg *PostgresDB) EventStreamDeleteByEventTypeID(ctx context.Context, eventTypeID int) (int, error) {
	row, err := pg.writer.ExecContext(ctx, eventStreamDeleteByEventTypeIDQuery, eventTypeID)
	if err != nil {
		return -1, errors.Wrap(err, "failed to delete from event streams")
	}

	countDeleted, err := row.RowsAffected()
	if err != nil {
		return -1, errors.Wrap(err, "could not count rows deleted")
	}

	return int(countDeleted), err
}

func (pg *PostgresDB) EventStreams(ctx context.Context) ([]*db.EventStream, error) {
	var eventStreams []*db.EventStream
	err := pg.reader.SelectContext(ctx, &eventStreams, eventStreamAllQuery)
	return eventStreams, err
}

func (pg *PostgresDB) EventStreamByID(ctx context.Context, id int) (*db.EventStream, error) {
	var eventStream db.EventStream
	err := pg.reader.GetContext(ctx, &eventStream, eventStreamByIDQuery, id)
	if err == sql.ErrNoRows {
		return nil, pgError(err, EventStreamsTableName)
	}
	return &eventStream, err
}

func (pg *PostgresDB) EventStreamsByEnvironment(ctx context.Context, environment string) ([]*db.EventStream, error) {
	var eventStreams []*db.EventStream
	err := pg.reader.SelectContext(ctx, &eventStreams, eventStreamWithEnvQuery, environment)
	return eventStreams, err
}

func (pg *PostgresDB) EventStreamsByName(ctx context.Context, name string) ([]*db.EventStream, error) {
	var eventStreams []*db.EventStream
	err := pg.reader.SelectContext(ctx, &eventStreams, eventStreamWithNameQuery, name)
	return eventStreams, err
}

func (pg *PostgresDB) EventStreamsByEventTypeID(ctx context.Context, eventTypeID int) ([]*db.EventStream, error) {
	var eventStreams []*db.EventStream
	err := pg.reader.SelectContext(ctx, &eventStreams, eventStreamByEventTypeIDQuery, eventTypeID)
	return eventStreams, err
}

func (pg *PostgresDB) EventStreamByNameAndEnvironment(ctx context.Context, name, environment string) (*db.EventStream, error) {
	var eventStreams []*db.EventStream
	err := pg.reader.SelectContext(ctx, &eventStreams, eventStreamWithNameAndEnvQuery, environment, name)
	if err != nil {
		return nil, err
	}
	if len(eventStreams) == 0 {
		return nil, pgError(sql.ErrNoRows, EventStreamsTableName)
	}
	return eventStreams[0], nil
}

func (pg *PostgresDB) EventStreamAcquireLease(ctx context.Context, resourceID int, timeout time.Duration) (db.AWSLease, context.Context, error) {
	return pg.Acquire(ctx, resourceID, EventStreamsTableName, timeout)
}

func (pg *PostgresDB) EventStreamReleaseLease(lease db.AWSLease) error {
	return lease.Release()
}
