package persistence

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"time"

	apperrors "github.com/nexuscrm/backend/pkg/errors"
	"github.com/nexuscrm/backend/pkg/query"
	"github.com/nexuscrm/backend/pkg/utils"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

// ErrOutboxEventAlreadyClaimed is returned by ClaimEvent when another worker
// has already transitioned the row out of the 'pending' state. Callers must
// treat this as a benign skip, not a failure.
var ErrOutboxEventAlreadyClaimed = errors.New("outbox event already claimed")

// outboxStatusProcessing is the transient status written by ClaimEvent when
// a pending row is successfully locked for dispatch. Defined locally rather
// than in shared/pkg/constants so this bug fix stays scoped to the outbox
// module; it is only ever written/read inside this file and its paired
// service. The column is VARCHAR so the value round-trips unchanged.
const outboxStatusProcessing = "processing"

// OutboxStaleProcessingSeconds bounds how long a row may sit in
// Status='processing' before GetPendingEvents / ClaimEvent are allowed to
// reclaim it. A worker that holds a row in 'processing' past this window
// is assumed dead (SIGKILL, panic before IncrementRetry, OOM, etc.) —
// otherwise the row would be stranded forever and the event never
// dispatched. 300s is chosen so a well-behaved handler has ample headroom
// above the default publish timeout + per-event retry backoff, while still
// recovering within one sentry poll interval (10 min).
const OutboxStaleProcessingSeconds = 300

// OutboxRepository handles database operations for the outbox pattern
type OutboxRepository struct {
	db *sql.DB
}

// NewOutboxRepository creates a new OutboxRepository
func NewOutboxRepository(db *sql.DB) *OutboxRepository {
	return &OutboxRepository{db: db}
}

// GetExecutor returns the transactional executor if present in context, or the DB connection
func (r *OutboxRepository) GetExecutor(ctx context.Context) Executor {
	if tx, ok := ctx.Value(txContextKey{}).(*sql.Tx); ok && tx != nil {
		return tx
	}
	return r.db
}

// Enqueue inserts a new event into the outbox
func (r *OutboxRepository) Enqueue(ctx context.Context, eventType string, payload interface{}) (string, error) {
	executor := r.GetExecutor(ctx)
	id := utils.GenerateID()

	payloadJSON, err := json.Marshal(payload)
	if err != nil {
		return "", apperrors.NewInternalError("failed to marshal event payload", err)
	}

	// Status defaults to 'pending'
	status := constants.OutboxStatusPending

	qr := query.Insert(constants.TableOutboxEvent, map[string]interface{}{
		constants.FieldID:                        id,
		constants.FieldSysOutboxEvent_EventType:  eventType,
		constants.FieldSysOutboxEvent_Payload:    payloadJSON,
		constants.FieldSysOutboxEvent_Status:     status,
		constants.FieldSysOutboxEvent_RetryCount: 0,
		constants.FieldCreatedDate:               time.Now(),
		constants.FieldLastModifiedDate:          time.Now(),
	}).Build()

	_, err = executor.ExecContext(ctx, qr.SQL, qr.Params...)
	if err != nil {
		return "", apperrors.NewInternalError("failed to enqueue event", err)
	}

	return id, nil
}

// GetPendingEvents retrieves IDs of pending events ordered by creation time.
//
// Rows stuck in Status='processing' past OutboxStaleProcessingSeconds are
// included in the result set and reclaimable: a worker that dies between
// ClaimEvent's UPDATE and the subsequent UpdateStatus / IncrementRetry
// would otherwise strand the row forever, invisible to the poll loop.
// The ClaimEvent predicate mirrors this so only one worker wins the
// reclaim race.
func (r *OutboxRepository) GetPendingEvents(ctx context.Context, limit int) ([]models.OutboxEvent, error) {
	cols := []string{
		constants.FieldID,
		constants.FieldSysOutboxEvent_EventType,
		constants.FieldSysOutboxEvent_Payload,
		constants.FieldSysOutboxEvent_RetryCount,
	}

	staleCutoff := time.Now().Add(-time.Duration(OutboxStaleProcessingSeconds) * time.Second)
	qr := query.From(constants.TableOutboxEvent).
		Select(cols).
		WhereRaw(
			"(`Status` = ? OR (`Status` = ? AND `LastModifiedDate` < ?))",
			[]interface{}{
				constants.OutboxStatusPending,
				outboxStatusProcessing,
				staleCutoff,
			},
		).
		OrderBy(constants.FieldCreatedDate, "ASC").
		Limit(limit).
		Build()

	rows, err := r.db.QueryContext(ctx, qr.SQL, qr.Params...)
	if err != nil {
		return nil, apperrors.NewInternalError("failed to query pending events", err)
	}
	defer rows.Close()

	var events []models.OutboxEvent
	for rows.Next() {
		var e models.OutboxEvent

		if err := rows.Scan(&e.ID, &e.EventType, &e.Payload, &e.RetryCount); err != nil {
			log.Printf("Warning: failed to scan outbox event: %v", err)
			continue
		}
		events = append(events, e)
	}
	if err := rows.Err(); err != nil {
		return nil, apperrors.NewInternalError("error iterating outbox events", err)
	}

	return events, nil
}

// ClaimEvent atomically transitions a pending (or stale 'processing') event
// to 'processing' so only one worker dispatches it.
//
// The original implementation used SELECT ... FOR UPDATE SKIP LOCKED, which
// silently degrades to a non-blocking read under TiDB optimistic txn mode
// (the cluster default on some TiDB Cloud Serverless branches; see
// backend/internal/features/automation/autonumber/reserve.go for the
// precedent fix). In optimistic mode two concurrent workers could each
// believe they had exclusively claimed the row and double-dispatch.
//
// The UPDATE ... WHERE (Status='pending' OR stale-processing) form below
// takes an exclusive row lock in BOTH txn modes. Of the N concurrent
// callers, exactly one sees RowsAffected()==1; the others see 0 because
// the WHERE clause no longer matches once the first UPDATE commits (or
// blocks until it does under pessimistic mode). The sentinel
// ErrOutboxEventAlreadyClaimed signals the benign-skip case so the
// service layer can distinguish it from a genuine DB error.
//
// The stale-processing predicate mirrors GetPendingEvents so a row whose
// previous worker died between claim and status-finalise becomes
// reclaimable after OutboxStaleProcessingSeconds — no separate sweeper
// goroutine is needed. LastModifiedDate is updated on every successful
// claim so a long-running-but-alive dispatch (unlikely; each event has a
// short publish timeout) does NOT get its row reclaimed mid-flight.
func (r *OutboxRepository) ClaimEvent(ctx context.Context, id string) (string, error) {
	staleCutoff := time.Now().Add(-time.Duration(OutboxStaleProcessingSeconds) * time.Second)
	qr := query.Update(constants.TableOutboxEvent).
		Set(map[string]interface{}{
			constants.FieldSysOutboxEvent_Status: outboxStatusProcessing,
			constants.FieldLastModifiedDate:      time.Now(),
		}).
		Where(constants.FieldID, id).
		WhereRaw(
			"(`Status` = ? OR (`Status` = ? AND `LastModifiedDate` < ?))",
			[]interface{}{
				constants.OutboxStatusPending,
				outboxStatusProcessing,
				staleCutoff,
			},
		).
		Build()

	exec := r.GetExecutor(ctx)
	res, err := exec.ExecContext(ctx, qr.SQL, qr.Params...)
	if err != nil {
		return "", err
	}
	n, err := res.RowsAffected()
	if err != nil {
		return "", err
	}
	if n == 0 {
		return "", ErrOutboxEventAlreadyClaimed
	}
	return id, nil
}

// UpdateStatus updates the status and related fields of an event
func (r *OutboxRepository) UpdateStatus(ctx context.Context, id string, status string, errMessage string) error {
	qb := query.Update(constants.TableOutboxEvent).
		Where(constants.FieldID, id)

	updates := map[string]interface{}{
		constants.FieldSysOutboxEvent_Status: status,
		constants.FieldLastModifiedDate:      time.Now(),
	}

	if status == string(constants.OutboxStatusProcessed) {
		updates[constants.FieldSysOutboxEvent_ProcessedDate] = time.Now()
	} else if status == string(constants.OutboxStatusFailed) {
		updates[constants.FieldSysOutboxEvent_ErrorMessage] = errMessage
	} else {
		return apperrors.NewValidationError("status", fmt.Sprintf("unsupported status update: %s", status))
	}

	qr := qb.Set(updates).Build()

	exec := r.GetExecutor(ctx)
	_, err := exec.ExecContext(ctx, qr.SQL, qr.Params...)
	return err
}

// IncrementRetry increments the retry count, updates the error message,
// and returns the row to 'pending' so the next poll picks it up. The row
// was moved to 'processing' by ClaimEvent; without this reset it would
// stay stuck and never retry.
func (r *OutboxRepository) IncrementRetry(ctx context.Context, id string, newCount int, errMessage string) error {
	qr := query.Update(constants.TableOutboxEvent).
		Set(map[string]interface{}{
			constants.FieldSysOutboxEvent_Status:       constants.OutboxStatusPending,
			constants.FieldSysOutboxEvent_RetryCount:   newCount,
			constants.FieldSysOutboxEvent_ErrorMessage: errMessage,
			constants.FieldLastModifiedDate:            time.Now(),
		}).
		Where(constants.FieldID, id).
		Build()

	exec := r.GetExecutor(ctx)
	_, err := exec.ExecContext(ctx, qr.SQL, qr.Params...)
	return err
}

// CleanupProcessed deletes old processed events
func (r *OutboxRepository) CleanupProcessed(ctx context.Context, cutoff time.Time) (int64, error) {
	qr := query.Delete(constants.TableOutboxEvent).
		Where(constants.FieldSysOutboxEvent_Status, constants.OutboxStatusProcessed).
		Where(constants.FieldSysOutboxEvent_ProcessedDate, cutoff, "<").
		Build()

	result, err := r.db.ExecContext(ctx, qr.SQL, qr.Params...)
	if err != nil {
		return 0, err
	}
	return result.RowsAffected()
}