package clickhouse

import (
	"context"
	"time"

	"github.com/flexprice/flexprice/internal/clickhouse"
	"github.com/flexprice/flexprice/internal/domain/events"
	ierr "github.com/flexprice/flexprice/internal/errors"
	"github.com/flexprice/flexprice/internal/logger"
	"github.com/flexprice/flexprice/internal/types"
)

type RawEventRepository struct {
	store  *clickhouse.ClickHouseStore
	logger *logger.Logger
}

func NewRawEventRepository(store *clickhouse.ClickHouseStore, logger *logger.Logger) events.RawEventRepository {
	return &RawEventRepository{store: store, logger: logger}
}

// FindRawEvents finds raw events with filtering and keyset pagination
// Query is optimized for the table structure:
// - PRIMARY KEY: (tenant_id, environment_id, external_customer_id, timestamp)
// - ORDER BY: (tenant_id, environment_id, external_customer_id, timestamp, event_name, id)
// - PARTITION BY: toYYYYMMDD(timestamp)
// - ENGINE: ReplacingMergeTree(version)
func (r *RawEventRepository) FindRawEvents(ctx context.Context, params *events.FindRawEventsParams) ([]*events.RawEvent, error) {
	span := StartRepositorySpan(ctx, "raw_event", "find_raw_events", map[string]interface{}{
		"batch_size":            params.BatchSize,
		"external_customer_ids": params.ExternalCustomerIDs,
	})
	defer FinishSpan(span)

	// Get tenant and environment ID from context
	tenantID := types.GetTenantID(ctx)
	environmentID := types.GetEnvironmentID(ctx)

	// Build query with filters following primary key order for optimal index usage
	query := `
		SELECT 
			id, tenant_id, environment_id, external_customer_id, event_name, 
			source, payload, field1, field2, field3, field4, field5, 
			field6, field7, field8, field9, field10, timestamp, ingested_at, 
			version, sign
		FROM raw_events
		WHERE tenant_id = ?
		AND environment_id = ?
	`

	args := []interface{}{tenantID, environmentID}

	// Add filters if provided - order matters for index usage
	// Follow the primary key order: tenant_id, environment_id, external_customer_id, timestamp
	if len(params.ExternalCustomerIDs) > 0 {
		query += " AND external_customer_id IN ?"
		args = append(args, params.ExternalCustomerIDs)
	}

	if !params.StartTime.IsZero() {
		query += " AND timestamp >= ?"
		args = append(args, params.StartTime)
	}

	if !params.EndTime.IsZero() {
		query += " AND timestamp <= ?"
		args = append(args, params.EndTime)
	}

	if len(params.EventNames) > 0 {
		query += " AND event_name IN ?"
		args = append(args, params.EventNames)
	}

	if len(params.EventIDs) > 0 {
		query += " AND id IN ?"
		args = append(args, params.EventIDs)
	}

	// Add sorting for consistent ordering
	query += " ORDER BY timestamp DESC, event_name DESC, id DESC"

	// Add OFFSET and LIMIT for simple pagination
	if params.Offset > 0 {
		query += " LIMIT ? OFFSET ?"
		if params.BatchSize > 0 {
			args = append(args, params.BatchSize, params.Offset)
		} else {
			args = append(args, 1000, params.Offset)
		}
	} else {
		// No offset, just limit
		if params.BatchSize > 0 {
			query += " LIMIT ?"
			args = append(args, params.BatchSize)
		} else {
			query += " LIMIT 1000"
		}
	}

	r.logger.Debug(ctx, "executing find raw events query",
		"query", query,
		"args", args,
		"external_customer_ids", params.ExternalCustomerIDs,
		"event_names", params.EventNames,
		"batch_size", params.BatchSize,
		"offset", params.Offset,
	)

	// Execute the query
	rows, err := r.store.GetConn().Query(ctx, query, args...)
	if err != nil {
		SetSpanError(span, err)
		return nil, ierr.WithError(err).
			WithHint("Failed to query raw events").
			Mark(ierr.ErrDatabase)
	}
	defer rows.Close()

	var eventsList []*events.RawEvent
	for rows.Next() {
		var event events.RawEvent

		err := rows.Scan(
			&event.ID,
			&event.TenantID,
			&event.EnvironmentID,
			&event.ExternalCustomerID,
			&event.EventName,
			&event.Source,
			&event.Payload,
			&event.Field1,
			&event.Field2,
			&event.Field3,
			&event.Field4,
			&event.Field5,
			&event.Field6,
			&event.Field7,
			&event.Field8,
			&event.Field9,
			&event.Field10,
			&event.Timestamp,
			&event.IngestedAt,
			&event.Version,
			&event.Sign,
		)
		if err != nil {
			SetSpanError(span, err)
			return nil, ierr.WithError(err).
				WithHint("Failed to scan raw event").
				Mark(ierr.ErrDatabase)
		}

		eventsList = append(eventsList, &event)
	}

	// Check for errors that occurred during iteration
	if err := rows.Err(); err != nil {
		SetSpanError(span, err)
		return nil, ierr.WithError(err).
			WithHint("Error occurred during row iteration").
			Mark(ierr.ErrDatabase)
	}

	r.logger.Debug(ctx, "fetched raw events from clickhouse",
		"count", len(eventsList),
		"expected_batch_size", params.BatchSize,
		"offset", params.Offset,
	)

	SetSpanSuccess(span)
	return eventsList, nil
}

// FindUnprocessedRawEvents finds raw events that haven't been processed yet.
//
// Strategy (2-step batch, keyset pagination):
//
//	Step A — Narrow scan: LEFT ANTI JOIN to find only unprocessed IDs.
//	         Reads 3 columns (timestamp, id, event_name) — cheap columnar I/O.
//	         Uses keyset pagination (timestamp, id) < (last_ts, last_id) instead
//	         of OFFSET, so every page costs the same regardless of depth.
//
//	Step B — Wide fetch: fetch all 20 columns only for the batch of IDs found
//	         in Step A. Bounded to BatchSize rows, never scans the full range.
//
// Caller drives pagination by passing params.KeysetCursor (nil = first batch).
// When the returned cursor is nil, there are no more batches.
//
// Performance note on the ANTI JOIN subquery:
//
//	When ExternalCustomerIDs is provided, the same filter is pushed into the
//	events subquery so ClickHouse builds a small per-customer hash table
//	instead of loading all tenant events. Benchmarked: 34s → <1s.
//
//	events table PRIMARY KEY is (tenant_id, environment_id) so external_customer_id
//	is NOT in the index — but filtering it inside the subquery still drastically
//	reduces the hash table size before the ANTI JOIN executes.
func (r *RawEventRepository) FindUnprocessedRawEvents(
	ctx context.Context,
	params *events.FindRawEventsParams,
) ([]*events.RawEvent, *events.KeysetCursor, error) {
	span := StartRepositorySpan(ctx, "raw_event", "find_unprocessed_raw_events", map[string]interface{}{
		"batch_size":            params.BatchSize,
		"external_customer_ids": params.ExternalCustomerIDs,
		"has_cursor":            params.KeysetCursor != nil,
	})
	defer FinishSpan(span)

	tenantID := types.GetTenantID(ctx)
	environmentID := types.GetEnvironmentID(ctx)

	batchSize := params.BatchSize
	if batchSize <= 0 {
		batchSize = 1000
	}

	// ── Step A: find unprocessed IDs (narrow scan) ───────────────────────────
	//
	// The events subquery is built dynamically based on filters:
	//
	//   Case 1 — ExternalCustomerIDs provided:
	//     Push the customer filter INTO the subquery so ClickHouse builds a
	//     small per-customer hash table instead of loading every event for the
	//     entire tenant. Critical optimization — verified 34s → <1s.
	//
	//   Case 2 — No customer filter (tenant-wide):
	//     Subquery covers all tenant+environment+timestamp events. Hash table
	//     is larger but acceptable — tenant-wide reprocessing is a rare,
	//     intentional operation.

	// Build subquery dynamically.
	subquery := `
		SELECT id
		FROM events
		WHERE tenant_id      = ?
		  AND environment_id = ?
		  AND timestamp >= ?
		  AND timestamp <  ?
	`
	subArgs := []interface{}{tenantID, environmentID, params.StartTime, params.EndTime}

	// Push customer filter into subquery when present — this is the key fix.
	// Narrows the ANTI JOIN hash table from "all tenant events" to
	// "this customer's events only".
	if len(params.ExternalCustomerIDs) > 0 {
		subquery += " AND external_customer_id IN ?"
		subArgs = append(subArgs, params.ExternalCustomerIDs)
	}

	stepAQuery := `
		SELECT r.timestamp, r.id, r.event_name
		FROM raw_events r
		LEFT ANTI JOIN (` + subquery + `) e ON r.id = e.id
		WHERE r.tenant_id      = ?
		  AND r.environment_id = ?
		  AND r.timestamp >= ?
		  AND r.timestamp <  ?
	`

	// Full args: subquery args first, then outer WHERE.
	stepAArgs := append(subArgs,
		tenantID, environmentID, params.StartTime, params.EndTime,
	)

	// Outer WHERE optional filters.
	// ExternalCustomerIDs first — matches raw_events primary key column order:
	// (tenant_id, environment_id, external_customer_id, timestamp, id)
	if len(params.ExternalCustomerIDs) > 0 {
		stepAQuery += " AND r.external_customer_id IN ?"
		stepAArgs = append(stepAArgs, params.ExternalCustomerIDs)
	}

	if len(params.EventNames) > 0 {
		stepAQuery += " AND r.event_name IN ?"
		stepAArgs = append(stepAArgs, params.EventNames)
	}

	if len(params.EventIDs) > 0 {
		stepAQuery += " AND r.id IN ?"
		stepAArgs = append(stepAArgs, params.EventIDs)
	}

	// Keyset pagination: tuple comparison lets ClickHouse use the sort key
	// directly. Replaces OFFSET which scans N*batchSize rows per page —
	// catastrophic at billions of events.
	if params.KeysetCursor != nil {
		stepAQuery += " AND (r.timestamp, r.id) < (?, ?)"
		stepAArgs = append(stepAArgs, params.KeysetCursor.LastTimestamp, params.KeysetCursor.LastID)
	}

	stepAQuery += " ORDER BY r.timestamp DESC, r.id DESC LIMIT ?"
	stepAArgs = append(stepAArgs, batchSize)

	r.logger.Debug(ctx, "executing step A: find unprocessed IDs",
		"tenant_id", tenantID,
		"external_customer_ids", params.ExternalCustomerIDs,
		"event_names", params.EventNames,
		"batch_size", batchSize,
		"has_cursor", params.KeysetCursor != nil,
		"customer_filter_in_subquery", len(params.ExternalCustomerIDs) > 0,
	)

	rowsA, err := r.store.GetConn().Query(ctx, stepAQuery, stepAArgs...)
	if err != nil {
		SetSpanError(span, err)
		return nil, nil, ierr.WithError(err).
			WithHint("Failed to query unprocessed event IDs (step A)").
			Mark(ierr.ErrDatabase)
	}
	defer rowsA.Close()

	var (
		ids        []string
		nextCursor *events.KeysetCursor
	)

	for rowsA.Next() {
		var (
			ts        time.Time
			id        string
			eventName string
		)
		if err := rowsA.Scan(&ts, &id, &eventName); err != nil {
			SetSpanError(span, err)
			return nil, nil, ierr.WithError(err).
				WithHint("Failed to scan unprocessed event ID (step A)").
				Mark(ierr.ErrDatabase)
		}
		ids = append(ids, id)
		// Updated each iteration — after the loop this holds the last (oldest)
		// row, which becomes the cursor for the next batch.
		nextCursor = &events.KeysetCursor{
			LastTimestamp: ts,
			LastID:        id,
		}
	}

	// rows.Err() catches truncated streams — network blips or server timeouts
	// that silently stop rows.Next() without returning a scan error.
	if err := rowsA.Err(); err != nil {
		SetSpanError(span, err)
		return nil, nil, ierr.WithError(err).
			WithHint("Row iteration error in step A").
			Mark(ierr.ErrDatabase)
	}

	r.logger.Debug(ctx, "step A complete",
		"unprocessed_id_count", len(ids),
		"has_next_cursor", nextCursor != nil,
	)

	if len(ids) == 0 {
		SetSpanSuccess(span)
		return nil, nil, nil
	}

	// Fewer rows than batchSize means the dataset is exhausted — no next cursor.
	if len(ids) < batchSize {
		nextCursor = nil
	}

	// ── Step B: fetch full rows for the IDs found above ──────────────────────
	//
	// Bounded to len(ids) rows (≤ batchSize). The 20-column wide read only
	// touches exactly the rows we know need processing.
	// tenant_id + environment_id prunes partitions; id IN narrows to exact rows.

	stepBQuery := `
		SELECT
			id, tenant_id, environment_id, external_customer_id, event_name,
			source, payload,
			field1, field2, field3, field4, field5,
			field6, field7, field8, field9, field10,
			timestamp, ingested_at, version, sign
		FROM raw_events
		WHERE tenant_id      = ?
		  AND environment_id = ?
		  AND id IN ?
		ORDER BY timestamp DESC, id DESC
	`
	stepBArgs := []interface{}{tenantID, environmentID, ids}

	r.logger.Debug(ctx, "executing step B: fetch full event rows",
		"id_count", len(ids),
	)

	rowsB, err := r.store.GetConn().Query(ctx, stepBQuery, stepBArgs...)
	if err != nil {
		SetSpanError(span, err)
		return nil, nil, ierr.WithError(err).
			WithHint("Failed to fetch full raw event rows (step B)").
			Mark(ierr.ErrDatabase)
	}
	defer rowsB.Close()

	eventsList := make([]*events.RawEvent, 0, len(ids))

	for rowsB.Next() {
		var event events.RawEvent
		if err := rowsB.Scan(
			&event.ID,
			&event.TenantID,
			&event.EnvironmentID,
			&event.ExternalCustomerID,
			&event.EventName,
			&event.Source,
			&event.Payload,
			&event.Field1,
			&event.Field2,
			&event.Field3,
			&event.Field4,
			&event.Field5,
			&event.Field6,
			&event.Field7,
			&event.Field8,
			&event.Field9,
			&event.Field10,
			&event.Timestamp,
			&event.IngestedAt,
			&event.Version,
			&event.Sign,
		); err != nil {
			SetSpanError(span, err)
			return nil, nil, ierr.WithError(err).
				WithHint("Failed to scan raw event row (step B)").
				Mark(ierr.ErrDatabase)
		}
		eventsList = append(eventsList, &event)
	}

	if err := rowsB.Err(); err != nil {
		SetSpanError(span, err)
		return nil, nil, ierr.WithError(err).
			WithHint("Row iteration error in step B").
			Mark(ierr.ErrDatabase)
	}

	r.logger.Debug(ctx, "unprocessed raw events batch",
		"count", len(eventsList),
		"has_next_batch", nextCursor != nil,
	)

	SetSpanSuccess(span)
	return eventsList, nextCursor, nil
}