package connector_events

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

func normalizeSourceSettleKey(key sourceSettleKey) sourceSettleKey {
	key.eventKind = strings.ToLower(strings.TrimSpace(key.eventKind))
	key.companionPack = strings.TrimSpace(key.companionPack)
	key.companionFlow = strings.TrimSpace(key.companionFlow)
	key.sourceKind = strings.TrimSpace(key.sourceKind)
	key.sourceID = strings.TrimSpace(key.sourceID)
	return key
}

// The map is local to the admitted transaction. SQL, not Go string equality,
// decides source identity under the stored columns' collation. SourceLaneKey
// cannot replace this identity: a route can override its serialization lane.
func loadSourceSettleDeadlines(ctx context.Context, tx *sql.Tx, now time.Time, deadlines map[sourceSettleKey]sql.NullTime) error {
	keys := make([]sourceSettleKey, 0, len(deadlines))
	for key := range deadlines {
		keys = append(keys, key)
	}
	const chunkSize = 256
	for offset := 0; offset < len(keys); offset += chunkSize {
		chunk := keys[offset:min(offset+chunkSize, len(keys))]
		query, args := sourceSettleBatchQuery(chunk, now)
		values := make([]sql.NullTime, len(chunk))
		dest := make([]any, len(chunk))
		for i := range values {
			dest[i] = &values[i]
		}
		if err := tx.QueryRowContext(ctx, query, args...).Scan(dest...); err != nil {
			return err
		}
		for i, value := range values {
			if value.Valid {
				deadlines[chunk[i]] = value
			}
		}
	}
	return nil
}

func sourceSettleBatchQuery(keys []sourceSettleKey, now time.Time) (string, []any) {
	inputs := make([]string, 0, len(keys))
	packs := make([]string, 0, 1)
	seenPacks := make(map[string]bool)
	args := make([]any, 0, len(keys)*8+1)
	for _, key := range keys {
		if !seenPacks[key.installedPackID] {
			seenPacks[key.installedPackID] = true
			packs = append(packs, key.installedPackID)
		}
		// Keep a temporal result for sql.NullTime without narrowing any stored
		// fractional seconds when reusing an open settle deadline.
		inputs = append(inputs, `CAST(MIN(CASE WHEN e.InstalledPackId=? AND e.EventKind=?
 AND COALESCE(e.RouteVersion,0)=? AND COALESCE(e.CompanionPackInstallId,'')=?
 AND COALESCE(e.CompanionFlowVersionId,'')=? AND e.SourceObjectKind=? AND e.SourceObjectID=?
 THEN e.NextAttemptAt END) AS DATETIME(6))`)
		args = append(args, key.installedPackID, key.eventKind, key.routeVersion, key.companionPack, key.companionFlow, key.sourceKind, key.sourceID)
	}
	args = append(args, now)
	for _, pack := range packs {
		args = append(args, pack)
	}
	// Keep requested values as parameters, not derived-table string columns:
	// those columns can acquire a competing collation and change equality.
	// Exclude unrelated packs without forcing a global deadline scan or a
	// pack-history scan. The optimizer can use the existing selective index.
	return `SELECT ` + strings.Join(inputs, ",") + `
 FROM _System_ConnectorEvent e
 WHERE e.Status='pending' AND e.RetryCount=0 AND e.NextAttemptAt>?
 AND e.InstalledPackId IN (` + sqlPlaceholders(len(packs)) + `)`, args
}