package records

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"hash"
	"io"
	"path"
	"strings"
	"time"

	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/kernel/pkg/utils"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/storage"
	"github.com/rs/zerolog/log"
)

type bulkIngestMetadata interface {
	GetSchema(ctx context.Context, objectAPIName string) *models.ObjectMetadata
}

type bulkIngestPermissions interface {
	CheckObjectPermissionWithUser(ctx context.Context, objectAPIName string, operation string, user *models.UserSession) bool
	RequiresRecordAwareCRUDDeferral(ctx context.Context, schema *models.ObjectMetadata, operation string) bool
	HasSuperUserCapability(ctx context.Context, user *models.UserSession) bool
}

type bulkIngestContract struct {
	jobs          *BulkService
	metadata      bulkIngestMetadata
	permissions   bulkIngestPermissions
	now           func() time.Time
	idgen         func() string
	objectStore   storage.ObjectStore
	storage       storage.Config
	storageDigest string
	storageErr    error
}

var _ contract.BulkIngest = (*bulkIngestContract)(nil)

func NewBulkIngestContract(jobs *BulkService, metadataSource bulkIngestMetadata, permissions bulkIngestPermissions, uploadDirOverride ...string) contract.BulkIngest {
	config := storage.LoadConfigFromEnv()
	if len(uploadDirOverride) > 0 {
		if override := strings.TrimSpace(uploadDirOverride[0]); override != "" {
			config.Provider = storage.ProviderFilesystem
			config.FilesystemRoot = override
			config.Bucket = storage.DefaultFilesystemBucket
		}
	}
	objectStore, storeErr := storage.NewStore(config)
	digest, digestErr := storage.ConfigAuthorityDigest(config)
	if storeErr == nil {
		storeErr = digestErr
	}
	return &bulkIngestContract{
		jobs:          jobs,
		metadata:      metadataSource,
		permissions:   permissions,
		now:           time.Now,
		idgen:         utils.GenerateID,
		objectStore:   objectStore,
		storage:       config,
		storageDigest: digest,
		storageErr:    storeErr,
	}
}

func (c *bulkIngestContract) CreateIngestJob(ctx context.Context, req contract.BulkIngestCreateRequest, user *models.UserSession) (*models.BulkJob, error) {
	if c == nil || c.jobs == nil || c.metadata == nil || c.permissions == nil {
		return nil, bulkIngestInternalError("bulk ingest service is not configured", nil)
	}
	schema := c.metadata.GetSchema(ctx, req.Object)
	if schema == nil {
		return nil, bulkIngestError(contract.BulkIngestErrorInvalidObject, fmt.Sprintf("Object '%s' does not exist", req.Object))
	}
	if strings.EqualFold(strings.TrimSpace(req.Object), constants.TableListView) {
		return nil, bulkIngestError(contract.BulkIngestErrorInvalidOperation, fmt.Sprintf("%s is not writable through bulk ingest", constants.TableListView))
	}

	requiredPerm, deferRecordAwareGate, err := c.requiredObjectPermission(ctx, schema, req.Operation)
	if err != nil {
		return nil, err
	}
	if req.Operation == models.OperationUpsert &&
		!c.permissions.CheckObjectPermissionWithUser(ctx, req.Object, constants.PermCreate, user) {
		return nil, bulkIngestError(contract.BulkIngestErrorInsufficientPermission, fmt.Sprintf("Upsert requires create permission on '%s'", req.Object))
	}
	if !deferRecordAwareGate && !c.permissions.CheckObjectPermissionWithUser(ctx, req.Object, requiredPerm, user) {
		return nil, bulkIngestError(contract.BulkIngestErrorInsufficientPermission, fmt.Sprintf("You do not have %s permission on '%s'", requiredPerm, req.Object))
	}

	now := c.now()
	job := &models.BulkJob{
		ID:                  c.idgen(),
		Object:              req.Object,
		Operation:           req.Operation,
		State:               models.JobStateOpen,
		ConcurrencyMode:     models.ConcurrencyParallel,
		ContentType:         "CSV",
		ColumnDelimiter:     "COMMA",
		LineEnding:          "LF",
		ApiVersion:          req.APIVersion,
		CreatedById:         bulkIngestUserID(user),
		CreatedDate:         now,
		SystemModstamp:      now,
		ExternalIdFieldName: req.ExternalIDFieldName,
	}
	if err := c.jobs.CreateJob(ctx, job); err != nil {
		return nil, bulkIngestInternalError("create bulk ingest job failed", err)
	}
	return job, nil
}

func (c *bulkIngestContract) GetIngestJob(ctx context.Context, jobID string, user *models.UserSession) (*models.BulkJob, error) {
	job, err := c.getJobForUser(ctx, jobID, user)
	if err != nil {
		return nil, err
	}
	return job, nil
}

func (c *bulkIngestContract) CloseIngestJob(ctx context.Context, jobID string, state models.BulkJobState, user *models.UserSession) (*models.BulkJob, error) {
	job, err := c.getJobForUser(ctx, jobID, user)
	if err != nil {
		return nil, err
	}
	if state != models.JobStateUploadComplete && state != models.JobStateAborted {
		return nil, bulkIngestError(contract.BulkIngestErrorInvalidRequest, "Only UploadComplete or Aborted state changes are allowed")
	}
	// A successful close is idempotent even if the worker later becomes
	// unavailable. Readiness gates only the admission of new durable work.
	if job.State == state {
		return job, nil
	}
	if state == models.JobStateUploadComplete {
		if err := c.jobs.CheckIngestWorkerReady(ctx); err != nil {
			return nil, bulkIngestInternalError("bulk ingest worker is not ready", err)
		}
	}
	var changed bool
	if state == models.JobStateUploadComplete {
		changed, err = c.jobs.AcceptIngestJob(ctx, jobID)
	} else {
		changed, err = c.jobs.AbortIngestJob(ctx, jobID)
	}
	if err != nil {
		return nil, bulkIngestInternalError("update bulk ingest job state failed", err)
	}
	if !changed && job.State != state {
		current, loadErr := c.jobs.GetJob(ctx, jobID)
		if loadErr != nil {
			return nil, bulkIngestInternalError("refresh bulk ingest job state failed", loadErr)
		}
		if current == nil || current.State != state {
			return nil, bulkIngestError(contract.BulkIngestErrorInvalidJobState, "Job cannot transition to the requested state")
		}
		job = current
	}
	updated, err := c.jobs.GetJob(ctx, jobID)
	if err != nil {
		log.Warn().Err(err).Str("job_id", jobID).Msg("Failed to refresh job after close")
		return job, nil
	}
	return updated, nil
}

func (c *bulkIngestContract) UploadIngestData(ctx context.Context, req contract.BulkIngestUploadRequest, user *models.UserSession) error {
	if req.Body == nil {
		return bulkIngestError(contract.BulkIngestErrorInvalidRequest, "Invalid request body")
	}
	job, err := c.getJobForUser(ctx, req.JobID, user)
	if err != nil {
		return err
	}
	if job.State != models.JobStateOpen {
		return bulkIngestError(contract.BulkIngestErrorInvalidJobState, "Job is not in Open state")
	}
	if req.ContentLength > contract.BulkIngestMaxUploadSizeBytes {
		return bulkIngestFileTooLargeError()
	}
	if c.storageErr != nil || c.objectStore == nil {
		return bulkIngestInternalError("bulk ingest object storage is unavailable", c.storageErr)
	}

	batchID := c.idgen()
	objectKey := path.Join("bulk-ingest", req.JobID, batchID+".csv")
	limited := newBulkUploadReader(req.Body, contract.BulkIngestMaxUploadSizeBytes)
	contentLength := req.ContentLength
	var length *int64
	if contentLength > 0 {
		length = &contentLength
	}
	stored, err := c.objectStore.PutObject(ctx, objectKey, limited, storage.PutObjectOptions{
		ContentType: "text/csv", ContentLength: length,
	})
	if err != nil {
		var bulkErr *contract.BulkIngestError
		if errors.As(err, &bulkErr) {
			return bulkErr
		}
		if errors.Is(err, errBulkUploadTooLarge) {
			return bulkIngestFileTooLargeError()
		}
		return bulkIngestInternalError("failed to save body to object storage", err)
	}
	if limited.size > contract.BulkIngestMaxUploadSizeBytes {
		_ = c.objectStore.DeleteObject(ctx, objectKey)
		return bulkIngestFileTooLargeError()
	}
	if stored.Key != objectKey || stored.Size != limited.size {
		_ = c.objectStore.DeleteObject(ctx, objectKey)
		return bulkIngestInternalError(
			"object storage returned inconsistent bulk upload identity",
			fmt.Errorf("key=%q size=%d, want key=%q size=%d", stored.Key, stored.Size, objectKey, limited.size),
		)
	}

	now := c.now()
	batch := &models.BulkBatch{
		ID:                     batchID,
		JobID:                  req.JobID,
		State:                  "Queued",
		ObjectKey:              stored.Key,
		StorageProvider:        c.storage.Provider,
		StorageBucket:          c.storage.Bucket,
		StorageAuthorityDigest: c.storageDigest,
		ContentSHA256:          limited.checksum(),
		ContentLength:          limited.size,
		CreatedDate:            now,
		SystemModstamp:         now,
	}
	admission, err := c.jobs.CreateBatchIfJobOpen(ctx, batch)
	if err != nil {
		if admission == ports.BulkBatchAdmissionNotCreated {
			deleteErr := c.objectStore.DeleteObject(ctx, objectKey)
			if deleteErr != nil {
				err = fmt.Errorf("%w (unaccepted object cleanup failed: %v)", err, deleteErr)
			}
		}
		return bulkIngestInternalError("create bulk ingest batch failed", err)
	}
	if admission == ports.BulkBatchAdmissionNotCreated {
		_ = c.objectStore.DeleteObject(ctx, objectKey)
		return bulkIngestError(contract.BulkIngestErrorInvalidJobState, "Job is not in Open state")
	}
	if admission != ports.BulkBatchAdmissionCreated {
		return bulkIngestInternalError(
			"create bulk ingest batch returned no durable outcome; uploaded object retained",
			errors.New("bulk batch admission outcome is unknown"),
		)
	}
	return nil
}

var errBulkUploadTooLarge = errors.New("bulk upload exceeds maximum size")

type bulkUploadReader struct {
	source io.Reader
	max    int64
	size   int64
	hash   hash.Hash
}

func newBulkUploadReader(source io.Reader, max int64) *bulkUploadReader {
	return &bulkUploadReader{source: source, max: max, hash: sha256.New()}
}

func (r *bulkUploadReader) Read(buffer []byte) (int, error) {
	if r.size >= r.max {
		var probe [1]byte
		n, err := r.source.Read(probe[:])
		if n > 0 {
			return 0, errBulkUploadTooLarge
		}
		return 0, err
	}
	remaining := r.max - r.size
	if int64(len(buffer)) > remaining {
		buffer = buffer[:remaining]
	}
	n, err := r.source.Read(buffer)
	if n > 0 {
		r.size += int64(n)
		_, _ = r.hash.Write(buffer[:n])
	}
	return n, err
}

func (r *bulkUploadReader) checksum() string { return hex.EncodeToString(r.hash.Sum(nil)) }

func (c *bulkIngestContract) requiredObjectPermission(ctx context.Context, schema *models.ObjectMetadata, operation models.BulkOperation) (string, bool, error) {
	switch operation {
	case models.OperationInsert:
		return constants.PermCreate, false, nil
	case models.OperationUpdate:
		return constants.PermEdit, c.shouldDeferObjectPermissionCheck(ctx, schema, operation), nil
	case models.OperationUpsert:
		return constants.PermEdit, c.shouldDeferObjectPermissionCheck(ctx, schema, operation), nil
	case models.OperationDelete:
		return constants.PermDelete, c.shouldDeferObjectPermissionCheck(ctx, schema, operation), nil
	default:
		return "", false, bulkIngestError(contract.BulkIngestErrorInvalidOperation, fmt.Sprintf("Invalid operation: '%s'. Valid operations: insert, update, upsert, delete", operation))
	}
}

func (c *bulkIngestContract) shouldDeferObjectPermissionCheck(ctx context.Context, schema *models.ObjectMetadata, operation models.BulkOperation) bool {
	if c == nil || c.permissions == nil || schema == nil {
		return false
	}
	var perm string
	switch operation {
	case models.OperationUpdate, models.OperationUpsert:
		perm = constants.PermEdit
	case models.OperationDelete:
		perm = constants.PermDelete
	default:
		return false
	}
	return c.permissions.RequiresRecordAwareCRUDDeferral(ctx, schema, perm)
}

func (c *bulkIngestContract) getJobForUser(ctx context.Context, jobID string, user *models.UserSession) (*models.BulkJob, error) {
	if c == nil || c.jobs == nil || c.permissions == nil {
		return nil, bulkIngestInternalError("bulk ingest service is not configured", nil)
	}
	job, err := c.jobs.GetJob(ctx, jobID)
	if err != nil {
		return nil, bulkIngestInternalError("get bulk ingest job failed", err)
	}
	if job == nil {
		return nil, bulkIngestError(contract.BulkIngestErrorNotFound, "Job not found")
	}
	if !job.Operation.IsIngest() {
		return nil, bulkIngestError(contract.BulkIngestErrorNotFound, "Job not found")
	}
	if !c.permissions.HasSuperUserCapability(ctx, user) && job.CreatedById != bulkIngestUserID(user) {
		return nil, bulkIngestError(contract.BulkIngestErrorAccessDenied, "Access denied to this job")
	}
	return job, nil
}

func bulkIngestError(kind contract.BulkIngestErrorKind, message string) *contract.BulkIngestError {
	return &contract.BulkIngestError{Kind: kind, Message: message}
}

func bulkIngestInternalError(message string, cause error) *contract.BulkIngestError {
	return &contract.BulkIngestError{Kind: contract.BulkIngestErrorInternal, Message: message, Cause: cause}
}

func bulkIngestFileTooLargeError() *contract.BulkIngestError {
	return bulkIngestError(contract.BulkIngestErrorRequestTooLarge, fmt.Sprintf("File size exceeds maximum allowed size of %d MB", contract.BulkIngestMaxUploadSizeBytes/(1024*1024)))
}

func bulkIngestUserID(user *models.UserSession) string {
	if user == nil {
		return ""
	}
	return user.ID
}