package persistence

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

	"github.com/google/uuid"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/shared/pkg/models"
)

var (
	ErrBulkClaimTransactionRolledBack = errors.New("bulk work claim transaction was proven rolled back")
	ErrBulkClaimFenced                = errors.New("bulk work claim was fenced")
	ErrBulkIngestJobAborted           = errors.New("bulk ingest job was aborted")
)

var bulkWorkRequiredJobColumns = map[string]string{
	"Id": "varchar(255)", "Operation": "varchar(255)", "State": "varchar(255)", "WorkBinding": "text",
	"AvailableAt": "datetime(6)", "ClaimOwner": "varchar(255)",
	"ClaimToken": "varchar(255)", "ClaimOwnershipEpoch": "decimal(20,0)",
	"LeaseExpiresAt": "datetime(6)", "RetryCount": "int", "LastFailure": "text",
}

var bulkWorkRequiredBatchColumns = map[string]string{
	"Id": "varchar(255)", "JobId": "varchar(255)", "State": "varchar(255)",
	"ObjectKey": "text", "StorageProvider": "varchar(32)",
	"StorageBucket": "varchar(255)", "StorageAuthorityDigest": "char(71)",
	"ContentSHA256": "char(64)", "ContentLength": "bigint",
	"NextRowOffset": "bigint", "LastError": "text",
}

// The shared BulkJob table also stores query/queryAll jobs. This worker owns
// only ingest mutations; every queue predicate uses this one closed SQL fact.
const bulkIngestOperationPredicate = "Operation IN ('insert', 'update', 'upsert', 'delete')"

type BulkWorkClaim struct {
	Job   *models.BulkJob
	Token string
}

type BulkWorkStore struct {
	db             *sql.DB
	repository     *BulkRepository
	transactions   *TransactionManager
	ownership      *RuntimeWorkOwnershipRepository
	holderID       string
	epoch          uint64
	initialBinding string
	lease          time.Duration
	maxAttempts    int
}

func NewBulkWorkStore(
	db *sql.DB,
	transactions *TransactionManager,
	ownership *RuntimeWorkOwnershipRepository,
	holderID string,
	epoch uint64,
	initialBinding string,
	lease time.Duration,
	maxAttempts int,
) (*BulkWorkStore, error) {
	if db == nil || transactions == nil || ownership == nil || holderID == "" || initialBinding == "" ||
		epoch == 0 || lease <= 0 || maxAttempts <= 0 {
		return nil, errors.New("bulk work store: database, ownership, holder, lease, and attempts are required")
	}
	return &BulkWorkStore{
		db: db, repository: NewBulkRepository(db, transactions), transactions: transactions,
		ownership: ownership, holderID: holderID, epoch: epoch, initialBinding: initialBinding, lease: lease,
		maxAttempts: maxAttempts,
	}, nil
}

func (s *BulkWorkStore) CheckStorageAuthority(
	ctx context.Context,
	provider, bucket, authorityDigest string,
) error {
	if s == nil {
		return errors.New("bulk work store: unavailable")
	}
	return VerifyPendingBulkStorageAuthority(ctx, s.db, provider, bucket, authorityDigest)
}

func (s *BulkWorkStore) CheckReady(ctx context.Context) error {
	if s == nil || s.db == nil {
		return errors.New("bulk work store: unavailable")
	}
	if err := VerifyRequiredTableColumns(ctx, s.db, "_System_BulkJob", bulkWorkRequiredJobColumns); err != nil {
		return fmt.Errorf("bulk work job schema: %w", err)
	}
	if err := VerifyRequiredTableColumns(ctx, s.db, "_System_BulkBatch", bulkWorkRequiredBatchColumns); err != nil {
		return fmt.Errorf("bulk work batch schema: %w", err)
	}
	if err := VerifyRequiredTableColumns(ctx, s.db, "_System_BulkWorkGeneration", map[string]string{
		"SingletonKey": "tinyint", "Generation": "decimal(20,0)",
	}); err != nil {
		return fmt.Errorf("bulk work generation schema: %w", err)
	}
	if err := VerifyExactTableIndex(ctx, s.db, "_System_BulkJob", "idx_bulk_job_available", false, []string{"State", "AvailableAt", "CreatedDate", "Id"}); err != nil {
		return fmt.Errorf("bulk work availability index: %w", err)
	}
	if err := VerifyNoLegacyPendingBulkWork(ctx, s.db); err != nil {
		return err
	}
	if err := s.ownership.CheckReady(ctx); err != nil {
		return err
	}
	return s.ownership.CheckFencesReady(ctx)
}

// VerifyNoLegacyPendingBulkWork blocks activation if the retired in-process
// processor left work that has no durable object/checkpoint identity. Open jobs
// with an uploaded batch are included because they can become accepted after
// activation without passing another migration gate.
// Silently skipping or guessing a legacy filesystem path would lose work or
// reintroduce a second storage authority. Operators must drain or explicitly
// resolve those rows before transferring the work family.
func VerifyNoLegacyPendingBulkWork(ctx context.Context, db *sql.DB) error {
	if db == nil {
		return errors.New("bulk work migration: database is required")
	}
	var count int64
	err := db.QueryRowContext(ctx, `SELECT COUNT(*)
FROM _System_BulkJob AS job
LEFT JOIN _System_BulkBatch AS batch ON batch.JobId = job.Id
	WHERE job.`+bulkIngestOperationPredicate+`
	  AND ((job.State IN ('UploadComplete', 'InProgress') AND job.AvailableAt IS NULL)
	   OR (job.State IN ('Open', 'UploadComplete', 'InProgress')
	    AND batch.Id IS NOT NULL AND (
	      batch.ObjectKey IS NULL OR batch.ObjectKey = ''
	      OR batch.StorageProvider IS NULL OR batch.StorageProvider = ''
	      OR batch.StorageBucket IS NULL OR batch.StorageBucket = ''
	      OR batch.StorageAuthorityDigest IS NULL OR batch.StorageAuthorityDigest = ''
	      OR batch.ContentSHA256 IS NULL OR batch.ContentSHA256 = '')))`).Scan(&count)
	if err != nil {
		return fmt.Errorf("bulk work migration: inspect accepted legacy work: %w", err)
	}
	if count != 0 {
		return fmt.Errorf("bulk work migration: %d accepted legacy job/batch rows lack durable worker identity; drain or resolve them before ownership activation", count)
	}
	return nil
}

// VerifyPendingBulkStorageAuthority makes a storage-location change an
// explicit deployment decision. It includes uploaded Open jobs because those
// jobs can be closed after activation and must still be readable by the new
// worker authority.
func VerifyPendingBulkStorageAuthority(
	ctx context.Context,
	db *sql.DB,
	provider, bucket, authorityDigest string,
) error {
	if db == nil || provider == "" || bucket == "" || authorityDigest == "" {
		return errors.New("bulk work storage authority: exact deployment authority is required")
	}
	var count int64
	err := db.QueryRowContext(ctx, `SELECT COUNT(*)
FROM _System_BulkJob AS job
JOIN _System_BulkBatch AS batch ON batch.JobId = job.Id
	WHERE job.`+bulkIngestOperationPredicate+`
  AND job.State IN ('Open', 'UploadComplete', 'InProgress')
  AND (batch.StorageProvider <> ? OR batch.StorageBucket <> ?
    OR batch.StorageAuthorityDigest <> ?)`, provider, bucket, authorityDigest).Scan(&count)
	if err != nil {
		return fmt.Errorf("bulk work storage authority: inspect accepted work: %w", err)
	}
	if count != 0 {
		return fmt.Errorf("bulk work storage authority: %d accepted batch rows belong to another storage authority; drain or migrate them before rollout", count)
	}
	return nil
}

func (s *BulkWorkStore) ReadGeneration(ctx context.Context) (uint64, error) {
	var generation uint64
	err := s.db.QueryRowContext(ctx, `SELECT Generation FROM _System_BulkWorkGeneration WHERE SingletonKey = 1`).Scan(&generation)
	if errors.Is(err, sql.ErrNoRows) {
		return 0, nil
	}
	if err != nil {
		return 0, fmt.Errorf("bulk work generation: %w", err)
	}
	return generation, nil
}

func (s *BulkWorkStore) Reconcile(ctx context.Context) error {
	return s.transactions.WithPessimisticRetry(ctx, func(txCtx context.Context) error {
		if err := s.ownership.LockActiveContext(txCtx); err != nil {
			return err
		}
		_, err := s.repository.executor(txCtx).ExecContext(txCtx, `UPDATE _System_BulkJob
SET State = 'Failed', ErrorMessage = 'bulk worker retry limit exceeded',
    LastFailure = 'bulk worker retry limit exceeded', AvailableAt = NULL,
    ClaimOwner = NULL, ClaimToken = NULL, ClaimOwnershipEpoch = NULL,
    LeaseExpiresAt = NULL, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE `+bulkIngestOperationPredicate+`
	AND State = 'InProgress' AND AvailableAt <= CURRENT_TIMESTAMP(6)
	AND RetryCount + 1 >= ?`, s.maxAttempts)
		return err
	}, 3)
}

func (s *BulkWorkStore) Claim(ctx context.Context, limit int) ([]BulkWorkClaim, bool, error) {
	if limit <= 0 {
		return nil, false, errors.New("bulk work claim: positive limit is required")
	}
	var claims []BulkWorkClaim
	more := false
	err := s.transactions.WithPessimisticRetry(ctx, func(txCtx context.Context) error {
		attemptClaims := make([]BulkWorkClaim, 0, limit)
		attemptMore := false
		if err := s.ownership.LockActiveContext(txCtx); err != nil {
			return err
		}
		exec := s.repository.executor(txCtx)
		rows, err := exec.QueryContext(txCtx, `SELECT Id FROM _System_BulkJob
WHERE `+bulkIngestOperationPredicate+`
	AND State IN ('UploadComplete', 'InProgress')
	AND AvailableAt <= CURRENT_TIMESTAMP(6)
	AND ((State = 'UploadComplete' AND RetryCount < ?)
	  OR (State = 'InProgress' AND RetryCount + 1 < ?))
ORDER BY AvailableAt, CreatedDate, Id
LIMIT ? FOR UPDATE SKIP LOCKED`, s.maxAttempts, s.maxAttempts, limit+1)
		if err != nil {
			return err
		}
		var ids []string
		for rows.Next() {
			var id string
			if err := rows.Scan(&id); err != nil {
				_ = rows.Close()
				return err
			}
			ids = append(ids, id)
		}
		if err := rows.Close(); err != nil {
			return err
		}
		if len(ids) > limit {
			attemptMore = true
			ids = ids[:limit]
		}
		for _, id := range ids {
			token := uuid.NewString()
			result, err := exec.ExecContext(txCtx, `UPDATE _System_BulkJob
SET State = 'InProgress', ClaimOwner = ?, ClaimToken = ?, ClaimOwnershipEpoch = ?,
    WorkBinding = CASE WHEN WorkBinding IS NULL OR WorkBinding = '' THEN ? ELSE WorkBinding END,
    RetryCount = RetryCount + CASE WHEN State = 'InProgress' THEN 1 ELSE 0 END,
    LeaseExpiresAt = TIMESTAMPADD(MICROSECOND, ?, CURRENT_TIMESTAMP(6)),
    AvailableAt = TIMESTAMPADD(MICROSECOND, ?, CURRENT_TIMESTAMP(6)),
    LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE Id = ? AND `+bulkIngestOperationPredicate+`
	AND State IN ('UploadComplete', 'InProgress')
  AND AvailableAt <= CURRENT_TIMESTAMP(6)`,
				s.holderID, token, s.epoch, s.initialBinding, s.lease.Microseconds(),
				s.lease.Microseconds(), id,
			)
			if err != nil {
				return err
			}
			affected, err := result.RowsAffected()
			if err != nil {
				return err
			}
			if affected != 1 {
				return errors.New("bulk work claim: locked candidate transition was lost")
			}
			job, err := s.repository.GetJob(txCtx, id)
			if err != nil || job == nil {
				if err == nil {
					err = errors.New("claimed bulk job disappeared")
				}
				return err
			}
			attemptClaims = append(attemptClaims, BulkWorkClaim{Job: job, Token: token})
		}
		// WithPessimisticRetry may replay this callback. Publish only the
		// complete candidate from the current attempt; failed attempts never
		// contribute claims or continuation state.
		claims = attemptClaims
		more = attemptMore
		return nil
	}, 3)
	if err != nil {
		return nil, false, err
	}
	return claims, more, err
}

func (s *BulkWorkStore) NextWorkAt(ctx context.Context) (time.Time, bool, error) {
	var next sql.NullTime
	err := s.db.QueryRowContext(ctx, `SELECT MIN(AvailableAt) FROM _System_BulkJob
WHERE `+bulkIngestOperationPredicate+`
	AND ((State = 'UploadComplete' AND RetryCount < ?)
	OR (State = 'InProgress' AND RetryCount + 1 < ?))`, s.maxAttempts, s.maxAttempts).Scan(&next)
	if err != nil {
		return time.Time{}, false, err
	}
	return next.Time, next.Valid, nil
}

func (s *BulkWorkStore) GetBatches(ctx context.Context, jobID string) ([]*models.BulkBatch, error) {
	return s.repository.GetBatches(ctx, jobID)
}

func (s *BulkWorkStore) RunClaimTransaction(
	ctx context.Context,
	claim BulkWorkClaim,
	fn func(context.Context, *BulkRepository) error,
) error {
	var completion ports.TransactionCompletionOutcome
	err := s.transactions.WithPessimisticRetry(ctx, func(txCtx context.Context) error {
		completion = ""
		ports.RegisterTransactionCompletionHook(txCtx, func(outcome ports.TransactionCompletionOutcome) {
			completion = outcome
		})
		return s.runClaimInActiveTransaction(txCtx, claim, fn)
	}, 3)
	if err != nil && completion == ports.TransactionCompletionRolledBack {
		return errors.Join(ErrBulkClaimTransactionRolledBack, err)
	}
	return err
}

// RunClaimInTransaction joins claim fencing, lease renewal, and a durable
// checkpoint to an existing Records-owned pessimistic transaction. It must not
// create a second transaction: the record engine owns lifecycle batching,
// derived-state maintenance, parent-lock replay, and the business effect.
func (s *BulkWorkStore) RunClaimInTransaction(
	ctx context.Context,
	claim BulkWorkClaim,
	fn func(context.Context, *BulkRepository) error,
) error {
	if s == nil || s.transactions == nil || !s.transactions.IsPessimisticTx(ctx) {
		return errors.New("bulk work claim join requires a Records-owned pessimistic transaction")
	}
	return s.runClaimInActiveTransaction(ctx, claim, fn)
}

func (s *BulkWorkStore) runClaimInActiveTransaction(
	ctx context.Context,
	claim BulkWorkClaim,
	fn func(context.Context, *BulkRepository) error,
) error {
	if fn == nil {
		return errors.New("bulk work claim transaction callback is required")
	}
	if err := s.ownership.LockActiveFenceContext(ctx, claim.Job.ID); err != nil {
		return err
	}
	exec := s.repository.executor(ctx)
	var id string
	if err := exec.QueryRowContext(ctx, `SELECT Id FROM _System_BulkJob
	WHERE Id = ? AND State = 'InProgress' AND ClaimOwner = ? AND ClaimToken = ?
	  AND ClaimOwnershipEpoch = ? FOR UPDATE`, claim.Job.ID, s.holderID, claim.Token, s.epoch).Scan(&id); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			var state models.BulkJobState
			stateErr := exec.QueryRowContext(ctx, `SELECT State FROM _System_BulkJob WHERE Id = ? FOR UPDATE`, claim.Job.ID).Scan(&state)
			if stateErr == nil && state == models.JobStateAborted {
				return ErrBulkIngestJobAborted
			}
			if stateErr != nil && !errors.Is(stateErr, sql.ErrNoRows) {
				return stateErr
			}
			return ErrBulkClaimFenced
		}
		return err
	}
	if err := fn(ctx, s.repository); err != nil {
		return err
	}
	_, err := exec.ExecContext(ctx, `UPDATE _System_BulkJob
SET LeaseExpiresAt = TIMESTAMPADD(MICROSECOND, ?, CURRENT_TIMESTAMP(6)),
    AvailableAt = TIMESTAMPADD(MICROSECOND, ?, CURRENT_TIMESTAMP(6)),
    LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE Id = ? AND ClaimToken = ?`, s.lease.Microseconds(), s.lease.Microseconds(), claim.Job.ID, claim.Token)
	return err
}

func (s *BulkWorkStore) UpdateBatchCheckpoint(
	ctx context.Context,
	batchID string,
	nextRow int64,
	processed, failed, total int,
	state, lastError string,
) error {
	result, err := s.repository.executor(ctx).ExecContext(ctx, `UPDATE _System_BulkBatch
SET NextRowOffset = ?, RecordsProcessed = ?, RecordsFailed = ?, TotalRecords = ?,
    State = ?, LastError = ?, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE Id = ?`, nextRow, processed, failed, total, state, bulkNullIfEmpty(lastError), batchID)
	if err != nil {
		return err
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return err
	}
	if affected != 1 {
		return errors.New("bulk work checkpoint: batch is missing")
	}
	return nil
}

func (s *BulkWorkStore) Complete(ctx context.Context, claim BulkWorkClaim, processed, failed int) error {
	return s.RunClaimTransaction(ctx, claim, func(txCtx context.Context, repository *BulkRepository) error {
		_, err := repository.executor(txCtx).ExecContext(txCtx, `UPDATE _System_BulkJob
SET State = 'JobComplete', NumberRecordsProcessed = ?, NumberRecordsFailed = ?,
    ErrorMessage = NULL, AvailableAt = NULL, ClaimOwner = NULL, ClaimToken = NULL,
    ClaimOwnershipEpoch = NULL, LeaseExpiresAt = NULL,
    LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE Id = ?`, processed, failed, claim.Job.ID)
		return err
	})
}

func (s *BulkWorkStore) Fail(ctx context.Context, claim BulkWorkClaim, cause error) error {
	const publicMessage = "bulk ingest processing failed"
	internalMessage := publicMessage
	if cause != nil {
		internalMessage = cause.Error()
	}
	return s.RunClaimTransaction(ctx, claim, func(txCtx context.Context, repository *BulkRepository) error {
		_, err := repository.executor(txCtx).ExecContext(txCtx, `UPDATE _System_BulkJob
SET State = 'Failed', ErrorMessage = ?, LastFailure = ?, AvailableAt = NULL,
    ClaimOwner = NULL, ClaimToken = NULL, ClaimOwnershipEpoch = NULL,
    LeaseExpiresAt = NULL, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE Id = ?`, publicMessage, internalMessage, claim.Job.ID)
		return err
	})
}

func (s *BulkWorkStore) Retry(ctx context.Context, claim BulkWorkClaim, cause error, delay time.Duration) error {
	message := "bulk work attempt failed"
	if cause != nil {
		message = cause.Error()
	}
	return s.RunClaimTransaction(ctx, claim, func(txCtx context.Context, repository *BulkRepository) error {
		_, err := repository.executor(txCtx).ExecContext(txCtx, `UPDATE _System_BulkJob
SET LastFailure = ?, AvailableAt = TIMESTAMPADD(MICROSECOND, ?, CURRENT_TIMESTAMP(6)),
    LeaseExpiresAt = NULL, ClaimOwner = NULL, ClaimToken = NULL,
    ClaimOwnershipEpoch = NULL, LastModifiedDate = CURRENT_TIMESTAMP(6)
WHERE Id = ?`, message, delay.Microseconds(), claim.Job.ID)
		return err
	})
}