package serviceactor

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

	"github.com/tidbcloud/AgentTeam/internal/coredb"
)

// Connection records a service actor connection to an external subject.
type Connection struct {
	ID                    string
	WorkspaceID           string
	ServiceDefinitionID   string
	ServiceInstallationID string
	ServiceKind           string
	ExternalSubjectRef    string
	DisplayLabel          string
	Credentials           []byte
	Config                []byte
	Status                string
	CreatedByMemberID     string
	CreatedAt             time.Time
	UpdatedAt             time.Time
	DisabledAt            *time.Time
}

// ConnectionUpsert describes a service connection write.
type ConnectionUpsert struct {
	ID                    string
	WorkspaceID           string
	ServiceDefinitionID   string
	ServiceInstallationID string
	ServiceKind           string
	ExternalSubjectRef    string
	DisplayLabel          string
	Credentials           []byte
	Config                []byte
	Status                string
	CreatedByMemberID     string
	Now                   time.Time
	DisabledAt            *time.Time
}

// UpsertConnection creates or updates a service connection.
func (store *Store) UpsertConnection(ctx context.Context, params ConnectionUpsert) (Connection, error) {
	execer, err := execerFromStore(store)
	if err != nil {
		return Connection{}, err
	}
	return upsertConnection(ctx, execer, params)
}

// UpsertConnection creates or updates a service connection in the transaction.
func (store *TxStore) UpsertConnection(ctx context.Context, params ConnectionUpsert) (Connection, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return Connection{}, err
	}
	return upsertConnection(ctx, execer, params)
}

// LoadConnection loads one service connection.
func (store *Store) LoadConnection(ctx context.Context, serviceConnectionID string) (Connection, error) {
	queryer, err := queryerFromStore(store)
	if err != nil {
		return Connection{}, err
	}
	return loadConnection(ctx, queryer, serviceConnectionID)
}

// ListActiveConnections lists active connections for a service installation.
func (store *Store) ListActiveConnections(ctx context.Context, serviceInstallationID string) ([]Connection, error) {
	queryer, err := queryerFromStore(store)
	if err != nil {
		return nil, err
	}
	return listActiveConnections(ctx, queryer, serviceInstallationID)
}

// ListActiveConnections lists active connections inside the caller-owned transaction.
func (store *TxStore) ListActiveConnections(ctx context.Context, serviceInstallationID string) ([]Connection, error) {
	queryer, err := queryerFromTxStore(store)
	if err != nil {
		return nil, err
	}
	return listActiveConnections(ctx, queryer, serviceInstallationID)
}

// ListActiveConnectionsByInstallationIDs lists active connections for the requested installations.
func (store *Store) ListActiveConnectionsByInstallationIDs(ctx context.Context, serviceInstallationIDs []string) (map[string][]Connection, error) {
	queryer, err := queryerFromStore(store)
	if err != nil {
		return nil, err
	}
	return listActiveConnectionsByInstallationIDs(ctx, queryer, serviceInstallationIDs)
}

// LoadActiveConnectionForUpdate locks an active connection for one installation/ref pair.
func (store *TxStore) LoadActiveConnectionForUpdate(ctx context.Context, serviceInstallationID string, externalSubjectRef string) (Connection, bool, error) {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return Connection{}, false, err
	}
	return loadActiveConnectionForUpdate(ctx, execer, serviceInstallationID, externalSubjectRef)
}

// DisableConnection disables one service connection.
func (store *Store) DisableConnection(ctx context.Context, serviceConnectionID string) error {
	execer, err := execerFromStore(store)
	if err != nil {
		return err
	}
	return disableConnection(ctx, execer, serviceConnectionID, time.Now().UTC())
}

// DisableConnectionsForInstallation disables active connections for one service installation.
func (store *TxStore) DisableConnectionsForInstallation(ctx context.Context, serviceInstallationID string, now time.Time) error {
	execer, err := execerFromTxStore(store)
	if err != nil {
		return err
	}
	return disableConnectionsForInstallation(ctx, execer, serviceInstallationID, now)
}

func upsertConnection(ctx context.Context, execer coredb.QueryExecer, params ConnectionUpsert) (Connection, error) {
	connectionID := trim(params.ID)
	workspaceID := trim(params.WorkspaceID)
	definitionID := trim(params.ServiceDefinitionID)
	installationID := trim(params.ServiceInstallationID)
	serviceKind := trim(params.ServiceKind)
	externalSubjectRef := trim(params.ExternalSubjectRef)
	createdByMemberID := trim(params.CreatedByMemberID)
	if connectionID == "" || workspaceID == "" || definitionID == "" || installationID == "" || serviceKind == "" || externalSubjectRef == "" || createdByMemberID == "" {
		return Connection{}, badRequest("service connection id, workspaceId, serviceDefinitionId, serviceInstallationId, kind, externalSubjectRef, and createdByMemberId are required")
	}
	if len(params.Credentials) == 0 {
		params.Credentials = []byte(`{}`)
	}
	status := trim(params.Status)
	if status == "" {
		status = StatusActive
	}
	now := params.Now.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	disabledAt := params.DisabledAt
	if status == StatusDisabled && disabledAt == nil {
		disabledAt = &now
	}
	if status == StatusActive {
		disabledAt = nil
	}
	if _, err := execer.Exec(ctx, `
INSERT INTO service_connections (
    id, workspace_id, service_definition_id, service_installation_id,
    service_kind, external_subject_ref, display_label, credentials, config,
    status, created_by_member_id, created_at, updated_at, disabled_at
) VALUES (
    $1, $2, $3, $4,
    $5, $6, $7, $8, $9,
    $10, $11, $12, $12, $13
)
ON DUPLICATE KEY UPDATE
    display_label = VALUES(display_label),
    credentials = VALUES(credentials),
    config = VALUES(config),
    status = VALUES(status),
    created_by_member_id = VALUES(created_by_member_id),
    updated_at = VALUES(updated_at),
    disabled_at = VALUES(disabled_at)
`, connectionID, workspaceID, definitionID, installationID, serviceKind, externalSubjectRef, trim(params.DisplayLabel), params.Credentials, params.Config, status, createdByMemberID, now, nullTimePtr(disabledAt)); err != nil {
		return Connection{}, fmt.Errorf("upsert service connection: %w", mapConnectionWriteError(err))
	}
	if status == StatusActive {
		connection, found, err := loadActiveConnection(ctx, execer, installationID, externalSubjectRef)
		if err != nil {
			return Connection{}, err
		}
		if found {
			return connection, nil
		}
	}
	return loadConnection(ctx, execer, connectionID)
}

func loadConnection(ctx context.Context, queryer coredb.QueryRower, serviceConnectionID string) (Connection, error) {
	row := queryer.QueryRow(ctx, connectionSelectSQL()+` WHERE id = $1`, trim(serviceConnectionID))
	connection, err := scanConnection(row)
	if err != nil {
		if coredb.IsNotFound(err) {
			return Connection{}, notFound("service connection not found")
		}
		return Connection{}, fmt.Errorf("load service connection: %w", err)
	}
	return connection, nil
}

func listActiveConnections(ctx context.Context, queryer coredb.Queryer, serviceInstallationID string) ([]Connection, error) {
	rows, err := queryer.Query(ctx, connectionSelectSQL()+`
WHERE service_installation_id = $1
  AND status = 'active'
ORDER BY created_at ASC, external_subject_ref ASC, id ASC
`, trim(serviceInstallationID))
	if err != nil {
		return nil, fmt.Errorf("list service connections: %w", err)
	}
	defer rows.Close()
	connections := make([]Connection, 0)
	for rows.Next() {
		connection, err := scanConnection(rows)
		if err != nil {
			return nil, fmt.Errorf("scan service connection: %w", err)
		}
		connections = append(connections, connection)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate service connections: %w", err)
	}
	return connections, nil
}

func listActiveConnectionsByInstallationIDs(ctx context.Context, queryer coredb.Queryer, serviceInstallationIDs []string) (map[string][]Connection, error) {
	installationIDs := trimSlice(serviceInstallationIDs)
	connectionsByInstallation := make(map[string][]Connection, len(installationIDs))
	for _, installationID := range installationIDs {
		connectionsByInstallation[installationID] = []Connection{}
	}
	if len(installationIDs) == 0 {
		return connectionsByInstallation, nil
	}
	requestedInstallationsCTE, requestedInstallationArgs, _, err := coredb.BuildStringValuesCTE("requested_service_installations", "service_installation_id", installationIDs, 1)
	if err != nil {
		return nil, fmt.Errorf("build service connections query: %w", err)
	}
	rows, err := queryer.Query(ctx, fmt.Sprintf(`
WITH %s
SELECT sc.id, sc.workspace_id, sc.service_definition_id, sc.service_installation_id,
       sc.service_kind, sc.external_subject_ref, sc.display_label, sc.credentials,
       sc.config, sc.status, sc.created_by_member_id, sc.created_at, sc.updated_at,
       sc.disabled_at
FROM requested_service_installations requested
JOIN service_connections sc
  ON sc.service_installation_id = requested.service_installation_id
 AND sc.status = 'active'
ORDER BY requested.service_installation_id ASC, sc.created_at ASC, sc.external_subject_ref ASC, sc.id ASC
`, requestedInstallationsCTE), requestedInstallationArgs...)
	if err != nil {
		return nil, fmt.Errorf("list service connections by installation ids: %w", err)
	}
	defer rows.Close()
	for rows.Next() {
		connection, err := scanConnection(rows)
		if err != nil {
			return nil, fmt.Errorf("scan service connection: %w", err)
		}
		connectionsByInstallation[connection.ServiceInstallationID] = append(connectionsByInstallation[connection.ServiceInstallationID], connection)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate service connections: %w", err)
	}
	return connectionsByInstallation, nil
}

func loadActiveConnectionForUpdate(ctx context.Context, queryer coredb.QueryRower, serviceInstallationID string, externalSubjectRef string) (Connection, bool, error) {
	row := queryer.QueryRow(ctx, connectionSelectSQL()+`
WHERE service_installation_id = $1
  AND external_subject_ref = $2
  AND status = 'active'
LIMIT 1
FOR UPDATE
`, trim(serviceInstallationID), trim(externalSubjectRef))
	connection, err := scanConnection(row)
	if err != nil {
		if coredb.IsNotFound(err) {
			return Connection{}, false, nil
		}
		return Connection{}, false, fmt.Errorf("load active service connection: %w", err)
	}
	return connection, true, nil
}

func loadActiveConnection(ctx context.Context, queryer coredb.QueryRower, serviceInstallationID string, externalSubjectRef string) (Connection, bool, error) {
	row := queryer.QueryRow(ctx, connectionSelectSQL()+`
WHERE service_installation_id = $1
  AND external_subject_ref = $2
  AND status = 'active'
LIMIT 1
`, trim(serviceInstallationID), trim(externalSubjectRef))
	connection, err := scanConnection(row)
	if err != nil {
		if coredb.IsNotFound(err) {
			return Connection{}, false, nil
		}
		return Connection{}, false, fmt.Errorf("load active service connection: %w", err)
	}
	return connection, true, nil
}

func disableConnection(ctx context.Context, execer coredb.QueryExecer, serviceConnectionID string, now time.Time) error {
	result, err := execer.Exec(ctx, `
UPDATE service_connections
SET status = 'disabled',
    disabled_at = $2,
    updated_at = $2
WHERE id = $1
`, trim(serviceConnectionID), now)
	if err != nil {
		return fmt.Errorf("disable service connection: %w", err)
	}
	if result.RowsAffected() == 0 {
		return notFound("service connection not found")
	}
	return nil
}

func disableConnectionsForInstallation(ctx context.Context, execer coredb.QueryExecer, serviceInstallationID string, now time.Time) error {
	now = now.UTC()
	if now.IsZero() {
		now = time.Now().UTC()
	}
	if _, err := execer.Exec(ctx, `
UPDATE service_connections
SET status = 'disabled',
    disabled_at = $2,
    updated_at = $2
WHERE service_installation_id = $1
  AND status = 'active'
`, trim(serviceInstallationID), now); err != nil {
		return fmt.Errorf("disable service connections for installation: %w", err)
	}
	return nil
}

func connectionSelectSQL() string {
	return `
SELECT id, workspace_id, service_definition_id, service_installation_id,
       service_kind, external_subject_ref, display_label, credentials, config,
       status, created_by_member_id, created_at, updated_at, disabled_at
FROM service_connections`
}

func scanConnection(row coredb.Row) (Connection, error) {
	var connection Connection
	var disabledAt sql.NullTime
	if err := row.Scan(
		&connection.ID,
		&connection.WorkspaceID,
		&connection.ServiceDefinitionID,
		&connection.ServiceInstallationID,
		&connection.ServiceKind,
		&connection.ExternalSubjectRef,
		&connection.DisplayLabel,
		&connection.Credentials,
		&connection.Config,
		&connection.Status,
		&connection.CreatedByMemberID,
		&connection.CreatedAt,
		&connection.UpdatedAt,
		&disabledAt,
	); err != nil {
		return Connection{}, err
	}
	connection.ID = trim(connection.ID)
	connection.WorkspaceID = trim(connection.WorkspaceID)
	connection.ServiceDefinitionID = trim(connection.ServiceDefinitionID)
	connection.ServiceInstallationID = trim(connection.ServiceInstallationID)
	connection.ServiceKind = trim(connection.ServiceKind)
	connection.ExternalSubjectRef = trim(connection.ExternalSubjectRef)
	connection.Status = trim(connection.Status)
	connection.DisabledAt = ptrFromNullTime(disabledAt)
	return connection, nil
}

func mapConnectionWriteError(err error) error {
	if constraint, ok := coredb.UniqueViolationConstraint(err); ok {
		switch constraint {
		case "service_connections_installation_ref_unique":
			return conflict("Connection already exists for this service installation.")
		case "service_connections_definition_workspace_ref_unique":
			return conflict("externalSubjectRef is already connected in this workspace.")
		}
	}
	return err
}