package persistence

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

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

// BulkRepository persists both the public Bulk API job shape and its durable
// execution lifecycle. Every method uses the caller's active transaction when
// present so a record effect and its checkpoint can share one commit boundary.
type BulkRepository struct {
	db           *sql.DB
	transactions *TransactionManager
}

func NewBulkRepository(db *sql.DB, transactions *TransactionManager) *BulkRepository {
	return &BulkRepository{db: db, transactions: transactions}
}

func (r *BulkRepository) executor(ctx context.Context) Executor {
	if r != nil && r.transactions != nil {
		if tx, ok := r.transactions.ExtractTx(ctx).(*sql.Tx); ok && tx != nil {
			return tx
		}
	}
	return r.db
}

func (r *BulkRepository) CreateJob(ctx context.Context, job *models.BulkJob) error {
	if r == nil || r.db == nil || job == nil {
		return errors.New("bulk repository: job and database are required")
	}
	_, err := r.executor(ctx).ExecContext(ctx, `INSERT INTO _System_BulkJob
(Id, Object, Operation, State, ConcurrencyMode, ContentType, ColumnDelimiter,
 LineEnding, ApiVersion, ExternalIdFieldName, CreatedById, CreatedDate,
 LastModifiedDate, JobType, ContentUrl, WorkBinding, AvailableAt, RetryCount)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		job.ID, job.Object, job.Operation, job.State, job.ConcurrencyMode,
		job.ContentType, job.ColumnDelimiter, job.LineEnding, job.ApiVersion,
		bulkNullIfEmpty(job.ExternalIdFieldName), bulkNullIfEmpty(job.CreatedById),
		job.CreatedDate, job.SystemModstamp, bulkNullIfEmpty(job.JobType),
		bulkNullIfEmpty(job.ContentUrl), bulkNullIfEmpty(job.WorkBinding), job.AvailableAt,
		job.RetryCount,
	)
	return err
}

func (r *BulkRepository) GetJob(ctx context.Context, jobID string) (*models.BulkJob, error) {
	if r == nil || r.db == nil {
		return nil, errors.New("bulk repository: database is required")
	}
	return r.scanJob(r.executor(ctx).QueryRowContext(ctx, `SELECT
 Id, Object, Operation, State, ConcurrencyMode, ContentType, ColumnDelimiter,
 LineEnding, ApiVersion, ExternalIdFieldName, CreatedById, CreatedDate,
 LastModifiedDate, ErrorMessage, NumberRecordsProcessed, NumberRecordsFailed,
 TotalProcessingTime, ApiActiveProcessingTime, ApexProcessingTime, JobType,
 ContentUrl, WorkBinding, AvailableAt, ClaimOwner, ClaimToken,
 CAST(ClaimOwnershipEpoch AS CHAR), LeaseExpiresAt, RetryCount, LastFailure
FROM _System_BulkJob WHERE Id = ?`, jobID))
}

func (r *BulkRepository) scanJob(row *sql.Row) (*models.BulkJob, error) {
	var job models.BulkJob
	var externalID, createdBy, errorMessage, jobType, contentURL, workBinding sql.NullString
	var availableAt, leaseExpiresAt sql.NullTime
	var claimOwner, claimToken, claimEpoch, lastFailure sql.NullString
	if err := row.Scan(
		&job.ID, &job.Object, &job.Operation, &job.State, &job.ConcurrencyMode,
		&job.ContentType, &job.ColumnDelimiter, &job.LineEnding, &job.ApiVersion,
		&externalID, &createdBy, &job.CreatedDate, &job.SystemModstamp,
		&errorMessage, &job.NumberRecordsProcessed, &job.NumberRecordsFailed,
		&job.TotalProcessingTime, &job.ApiActiveProcessingTime,
		&job.ApexProcessingTime, &jobType, &contentURL, &workBinding, &availableAt,
		&claimOwner, &claimToken, &claimEpoch, &leaseExpiresAt, &job.RetryCount,
		&lastFailure,
	); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil, nil
		}
		return nil, err
	}
	job.ExternalIdFieldName = externalID.String
	job.CreatedById = createdBy.String
	job.ErrorMessage = errorMessage.String
	job.JobType = jobType.String
	job.ContentUrl = contentURL.String
	job.WorkBinding = workBinding.String
	job.ClaimOwner = claimOwner.String
	job.ClaimToken = claimToken.String
	job.LastFailure = lastFailure.String
	if availableAt.Valid {
		value := availableAt.Time
		job.AvailableAt = &value
	}
	if leaseExpiresAt.Valid {
		value := leaseExpiresAt.Time
		job.LeaseExpiresAt = &value
	}
	if claimEpoch.Valid {
		if _, err := fmt.Sscan(claimEpoch.String, &job.ClaimOwnershipEpoch); err != nil {
			return nil, fmt.Errorf("bulk repository: invalid claim ownership epoch: %w", err)
		}
	}
	return &job, nil
}

func (r *BulkRepository) UpdateJob(ctx context.Context, job *models.BulkJob) error {
	if job == nil {
		return errors.New("bulk repository: job is required")
	}
	_, err := r.executor(ctx).ExecContext(ctx, `UPDATE _System_BulkJob SET
 State = ?, LastModifiedDate = CURRENT_TIMESTAMP(6), ErrorMessage = ?,
 JobType = ?, ContentUrl = ?, Object = ?, ExternalIdFieldName = ?,
 NumberRecordsProcessed = ?, NumberRecordsFailed = ?, TotalProcessingTime = ?,
 ApiActiveProcessingTime = ?, ApexProcessingTime = ?
WHERE Id = ?`,
		job.State, bulkNullIfEmpty(job.ErrorMessage), bulkNullIfEmpty(job.JobType),
		bulkNullIfEmpty(job.ContentUrl), job.Object, bulkNullIfEmpty(job.ExternalIdFieldName),
		job.NumberRecordsProcessed, job.NumberRecordsFailed, job.TotalProcessingTime,
		job.ApiActiveProcessingTime, job.ApexProcessingTime, job.ID,
	)
	return err
}

func (r *BulkRepository) insertBatch(ctx context.Context, exec Executor, batch *models.BulkBatch) error {
	_, err := exec.ExecContext(ctx, `INSERT INTO _System_BulkBatch
(Id, JobId, State, RecordsProcessed, RecordsFailed, TotalRecords, ObjectKey,
 StorageProvider, StorageBucket, StorageAuthorityDigest, ContentSHA256,
 ContentLength, NextRowOffset, LastError, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		batch.ID, batch.JobID, batch.State, batch.RecordsProcessed,
		batch.RecordsFailed, batch.TotalRecords, bulkNullIfEmpty(batch.ObjectKey),
		bulkNullIfEmpty(batch.StorageProvider), bulkNullIfEmpty(batch.StorageBucket),
		bulkNullIfEmpty(batch.StorageAuthorityDigest), bulkNullIfEmpty(batch.ContentSHA256),
		batch.ContentLength, batch.NextRowOffset, bulkNullIfEmpty(batch.LastError),
		batch.CreatedDate, batch.SystemModstamp,
	)
	return err
}

func (r *BulkRepository) GetBatches(ctx context.Context, jobID string) ([]*models.BulkBatch, error) {
	rows, err := r.executor(ctx).QueryContext(ctx, `SELECT
 Id, JobId, State, RecordsProcessed, RecordsFailed, TotalRecords, ObjectKey,
 StorageProvider, StorageBucket, StorageAuthorityDigest, ContentSHA256,
 ContentLength, NextRowOffset, LastError, CreatedDate, LastModifiedDate
FROM _System_BulkBatch WHERE JobId = ? ORDER BY CreatedDate, Id`, jobID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var batches []*models.BulkBatch
	for rows.Next() {
		var batch models.BulkBatch
		var objectKey, provider, bucket, authority, checksum, lastError sql.NullString
		if err := rows.Scan(
			&batch.ID, &batch.JobID, &batch.State, &batch.RecordsProcessed,
			&batch.RecordsFailed, &batch.TotalRecords, &objectKey, &provider,
			&bucket, &authority, &checksum, &batch.ContentLength,
			&batch.NextRowOffset, &lastError, &batch.CreatedDate,
			&batch.SystemModstamp,
		); err != nil {
			return nil, err
		}
		batch.ObjectKey = objectKey.String
		batch.StorageProvider = provider.String
		batch.StorageBucket = bucket.String
		batch.StorageAuthorityDigest = authority.String
		batch.ContentSHA256 = checksum.String
		batch.LastError = lastError.String
		batches = append(batches, &batch)
	}
	return batches, rows.Err()
}

// CreateBatchIfJobOpen serializes upload registration against close and returns
// a typed durability outcome. Callers may discard the uploaded object only
// when this method proves the batch was not created; commit/rollback
// acknowledgement errors remain outcome-unknown.
func (r *BulkRepository) CreateBatchIfJobOpen(ctx context.Context, batch *models.BulkBatch) (ports.BulkBatchAdmissionOutcome, error) {
	if r == nil || r.transactions == nil || batch == nil {
		return ports.BulkBatchAdmissionNotCreated, errors.New("bulk repository: transactional batch admission is unavailable")
	}
	if r.transactions.IsTransactionActive(ctx) {
		return ports.BulkBatchAdmissionUnknown, errors.New("bulk repository: batch admission must own its transaction outcome")
	}
	var completion ports.TransactionCompletionOutcome
	attemptStarted := false
	created := false
	err := r.transactions.WithPessimisticRetry(ctx, func(txCtx context.Context) error {
		attemptStarted = true
		created = false
		if !ports.RegisterTransactionCompletionHook(txCtx, func(outcome ports.TransactionCompletionOutcome) {
			completion = outcome
		}) {
			return errors.New("bulk repository: transaction completion outcome is unavailable")
		}
		exec := r.executor(txCtx)
		var state models.BulkJobState
		if err := exec.QueryRowContext(txCtx, `SELECT State FROM _System_BulkJob
WHERE Id = ? AND `+bulkIngestOperationPredicate+` FOR UPDATE`, batch.JobID).Scan(&state); err != nil {
			return err
		}
		if state != models.JobStateOpen {
			return nil
		}
		if err := r.insertBatch(txCtx, exec, batch); err != nil {
			return err
		}
		created = true
		return nil
	}, 3)
	if err == nil {
		if created {
			return ports.BulkBatchAdmissionCreated, nil
		}
		return ports.BulkBatchAdmissionNotCreated, nil
	}
	if completion == ports.TransactionCompletionRolledBack || !attemptStarted {
		return ports.BulkBatchAdmissionNotCreated, err
	}
	if completion == ports.TransactionCompletionCommitError && !created {
		return ports.BulkBatchAdmissionNotCreated, err
	}
	if completion == ports.TransactionCompletionCommitError && created {
		persisted, readErr := r.getBatchByID(ctx, batch.ID)
		if readErr == nil && bulkBatchAdmissionIdentityMatches(persisted, batch) {
			return ports.BulkBatchAdmissionCreated, nil
		}
		if readErr != nil {
			err = fmt.Errorf("%w (batch admission read-back failed: %v)", err, readErr)
		}
	}
	return ports.BulkBatchAdmissionUnknown, err
}

func (r *BulkRepository) getBatchByID(ctx context.Context, batchID string) (*models.BulkBatch, error) {
	var batch models.BulkBatch
	var objectKey, provider, bucket, authority, checksum sql.NullString
	err := r.db.QueryRowContext(ctx, `SELECT
 Id, JobId, State, ObjectKey, StorageProvider, StorageBucket,
 StorageAuthorityDigest, ContentSHA256, ContentLength
FROM _System_BulkBatch WHERE Id = ?`, batchID).Scan(
		&batch.ID, &batch.JobID, &batch.State, &objectKey, &provider, &bucket,
		&authority, &checksum, &batch.ContentLength,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	batch.ObjectKey = objectKey.String
	batch.StorageProvider = provider.String
	batch.StorageBucket = bucket.String
	batch.StorageAuthorityDigest = authority.String
	batch.ContentSHA256 = checksum.String
	return &batch, nil
}

func bulkBatchAdmissionIdentityMatches(persisted, requested *models.BulkBatch) bool {
	return persisted != nil && requested != nil &&
		persisted.ID == requested.ID && persisted.JobID == requested.JobID &&
		persisted.State == requested.State && persisted.ObjectKey == requested.ObjectKey &&
		persisted.StorageProvider == requested.StorageProvider &&
		persisted.StorageBucket == requested.StorageBucket &&
		persisted.StorageAuthorityDigest == requested.StorageAuthorityDigest &&
		persisted.ContentSHA256 == requested.ContentSHA256 &&
		persisted.ContentLength == requested.ContentLength
}

// AcceptIngestJob is the sole ingest acceptance transition. Making a job
// runnable and advancing the cheap wake cursor are one durable commit.
func (r *BulkRepository) AcceptIngestJob(ctx context.Context, jobID string) (bool, error) {
	if r == nil || r.transactions == nil {
		return false, errors.New("bulk repository: transactional state transition is unavailable")
	}
	changed := false
	err := r.transactions.WithPessimisticRetry(ctx, func(txCtx context.Context) error {
		// WithPessimisticRetry may replay this callback after a proven rollback.
		// Publish only the outcome of the attempt that actually commits.
		changed = false
		exec := r.executor(txCtx)
		result, err := exec.ExecContext(txCtx, `UPDATE _System_BulkJob
	SET State = 'UploadComplete', AvailableAt = CURRENT_TIMESTAMP(6),
	    ClaimOwner = NULL, ClaimToken = NULL,
	    ClaimOwnershipEpoch = NULL, LeaseExpiresAt = NULL,
	    LastModifiedDate = CURRENT_TIMESTAMP(6)
	WHERE Id = ? AND `+bulkIngestOperationPredicate+` AND State = 'Open'`, jobID)
		if err != nil {
			return err
		}
		affected, err := result.RowsAffected()
		if err != nil {
			return err
		}
		if affected == 0 {
			return nil
		}
		changed = true
		return advanceBulkWorkGeneration(txCtx, exec)
	}, 3)
	return changed, err
}

// AbortIngestJob owns the complete nonterminal abort transition. The same row
// update removes queued availability and invalidates any active claim. A
// Records effect/checkpoint transaction that already locked the job wins its
// current atomic chunk; otherwise this transition wins and the claim fence
// prevents any later chunk from starting. Previously committed rows are never
// compensated.
func (r *BulkRepository) AbortIngestJob(ctx context.Context, jobID string) (bool, error) {
	if r == nil || r.transactions == nil {
		return false, errors.New("bulk repository: transactional state transition is unavailable")
	}
	changed := false
	err := r.transactions.WithPessimisticRetry(ctx, func(txCtx context.Context) error {
		// WithPessimisticRetry may replay this callback after a proven rollback.
		changed = false
		result, err := r.executor(txCtx).ExecContext(txCtx, `UPDATE _System_BulkJob
	SET State = 'Aborted', AvailableAt = NULL,
	    ClaimOwner = NULL, ClaimToken = NULL, ClaimOwnershipEpoch = NULL,
	    LeaseExpiresAt = NULL, LastModifiedDate = CURRENT_TIMESTAMP(6)
	WHERE Id = ? AND `+bulkIngestOperationPredicate+` AND State IN ('Open', 'UploadComplete', 'InProgress')`, jobID)
		if err != nil {
			return err
		}
		affected, err := result.RowsAffected()
		if err != nil {
			return err
		}
		changed = affected == 1
		return nil
	}, 3)
	return changed, err
}

func (r *BulkRepository) VerifyActiveWorker(ctx context.Context, workFamily string) error {
	if r == nil || r.db == nil {
		return errors.New("bulk repository: database is required")
	}
	return VerifyActiveRuntimeWorkFamily(ctx, r.db, workFamily)
}

func advanceBulkWorkGeneration(ctx context.Context, exec Executor) error {
	_, err := exec.ExecContext(ctx, `INSERT INTO _System_BulkWorkGeneration
(SingletonKey, Generation) VALUES (1, 1)
ON DUPLICATE KEY UPDATE Generation = Generation + 1,
LastModifiedDate = CURRENT_TIMESTAMP(6)`)
	return err
}

func bulkNullIfEmpty(value string) interface{} {
	if value == "" {
		return nil
	}
	return value
}