package postgres

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

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

const (
	SubscriptionsTableName = "subscriptions"

	subscriptionSelectBase = `
SELECT * FROM subscriptions
`

	subscriptionCreateQuery = `
INSERT INTO subscriptions (status, error, subscription_target_id, event_stream_id, sns_subscription_arn)
VALUES (:status, :error, :subscription_target_id, :event_stream_id, '')
RETURNING id
`

	subscriptionUpdateInfraQuery = `
UPDATE subscriptions
SET error = $2, status = $3, sns_subscription_arn = $4
WHERE id = $1
`

	subscriptionAllQuery = subscriptionSelectBase + `
ORDER BY subscription_target_id, event_stream_id
`

	subscriptionByIDQuery = subscriptionSelectBase + `
WHERE id = $1
`

	subscriptionsByTargetIDQuery = subscriptionSelectBase + `
WHERE subscription_target_id = $1
ORDER BY event_stream_id
`

	subscriptionByEventStreamIDAndTargetIDQuery = subscriptionSelectBase + `
WHERE event_stream_id = $1 AND subscription_target_id = $2
LIMIT 1
`

	subscriptionsByEventStreamIDQuery = subscriptionSelectBase + `
WHERE event_stream_id = $1
ORDER BY subscription_target_id
`

	subscriptionDeleteQuery = `
DELETE FROM subscriptions
WHERE id = $1
`
)

func (pg *PostgresDB) SubscriptionCreate(ctx context.Context, subscription *db.Subscription) (int, error) {
	var id int
	namedQuery, err := pg.writer.PrepareNamedContext(ctx, subscriptionCreateQuery)
	if err != nil {
		return -1, err
	}

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

func (pg *PostgresDB) SubscriptionUpdateInfra(ctx context.Context, lease db.AWSLease, id int, subscriptionInfraUpdate *db.SubscriptionInfraUpdate) (int, error) {
	if lease == nil {
		return -1, errors.New("nil lease object passed to SubscriptionsTable.UpdateInfra")
	}

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

	res, err := pg.writer.ExecContext(ctx, subscriptionUpdateInfraQuery, id, subscriptionInfraUpdate.Error, subscriptionInfraUpdate.Status, subscriptionInfraUpdate.SNSSubscriptionARN)
	if err != nil {
		return -1, err
	}

	rowsAffected, err := res.RowsAffected()
	if err != nil {
		return -1, err
	}

	if rowsAffected == 0 {
		return -1, db.ErrResourceNotFound
	}

	return id, nil
}
func (pg *PostgresDB) SubscriptionDelete(ctx context.Context, lease db.AWSLease, subscriptionID int) error {
	if lease == nil {
		return errors.New("nil lease object passed to SubscriptionDeleteBySubscriptionTargetID")
	}

	if lease.Expired() {
		return db.ErrLeaseExpired
	}

	res, err := pg.writer.ExecContext(ctx, subscriptionDeleteQuery, subscriptionID)
	if err != nil {
		return err
	}

	rowsAffected, err := res.RowsAffected()
	if err != nil {
		return err
	}

	if rowsAffected == 0 {
		return db.ErrResourceNotFound
	}

	return nil
}

func (pg *PostgresDB) Subscriptions(ctx context.Context) ([]*db.Subscription, error) {
	var subs []*db.Subscription
	err := pg.reader.SelectContext(ctx, &subs, subscriptionAllQuery)
	return subs, err
}

func (pg *PostgresDB) SubscriptionByID(ctx context.Context, id int) (*db.Subscription, error) {
	var sub db.Subscription
	err := pg.reader.GetContext(ctx, &sub, subscriptionByIDQuery, id)
	if err == sql.ErrNoRows {
		return nil, db.ErrResourceNotFound
	}
	return &sub, err
}

// TODO: https://jira.twitch.com/browse/ASYNC-365
// Remove this db call once we add an id to subs twirp api
func (pg *PostgresDB) SubscriptionByEventStreamIDAndSubscriptionTargetID(ctx context.Context, eID, tID int) (*db.Subscription, error) {
	var sub db.Subscription
	err := pg.reader.GetContext(ctx, &sub, subscriptionByEventStreamIDAndTargetIDQuery, eID, tID)
	if err == sql.ErrNoRows {
		return nil, db.ErrResourceNotFound
	}
	return &sub, err
}

func (pg *PostgresDB) SubscriptionsBySubscriptionTargetID(ctx context.Context, subscriptionTargetID int) ([]*db.Subscription, error) {
	var subs []*db.Subscription
	err := pg.reader.SelectContext(ctx, &subs, subscriptionsByTargetIDQuery, subscriptionTargetID)
	return subs, err
}

func (pg *PostgresDB) SubscriptionsByEventStreamID(ctx context.Context, eventStreamID int) ([]*db.Subscription, error) {
	var subs []*db.Subscription
	err := pg.reader.SelectContext(ctx, &subs, subscriptionsByEventStreamIDQuery, eventStreamID)
	return subs, err
}

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

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