package sfcrmappservice

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"strconv"
	"strings"

	crmwire "github.com/pingcap-inc/nexuscrm-app-packs/sfcrm/wire"
	bulkingestsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/bulkingestv1"
	commonsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commonv1"
)

// Salesforce Bulk API 2.0 public route and wire contract:
// https://developer.salesforce.com/docs/atlas.en-us.254.0.api_asynch.meta/api_asynch/

type RuntimeCreateBulkIngestJobFunc func(context.Context, string, bulkingestsemantic.CreateRequest) (bulkingestsemantic.CreateResponse, error)
type RuntimeReadBulkIngestJobFunc func(context.Context, string, bulkingestsemantic.ReadRequest) (bulkingestsemantic.ReadResponse, error)
type RuntimeCloseBulkIngestJobFunc func(context.Context, string, bulkingestsemantic.CloseRequest) (bulkingestsemantic.CloseResponse, error)

type RuntimeBulkIngestUpload interface {
	Send([]byte) error
	CloseAndRecv() (bulkingestsemantic.UploadResponse, error)
	Cancel()
}

type RuntimeBeginBulkIngestUploadFunc func(context.Context, string, bulkingestsemantic.UploadHeader) (RuntimeBulkIngestUpload, error)

type RuntimeBulkIngestClient struct {
	create RuntimeCreateBulkIngestJobFunc
	read   RuntimeReadBulkIngestJobFunc
	close  RuntimeCloseBulkIngestJobFunc
	begin  RuntimeBeginBulkIngestUploadFunc
}

func NewRuntimeBulkIngestClient(create RuntimeCreateBulkIngestJobFunc, read RuntimeReadBulkIngestJobFunc, close RuntimeCloseBulkIngestJobFunc, begin RuntimeBeginBulkIngestUploadFunc) (*RuntimeBulkIngestClient, error) {
	if create == nil || read == nil || close == nil || begin == nil {
		return nil, errors.New("sfcrm AppService: complete Runtime Bulk Ingest lifecycle is required")
	}
	return &RuntimeBulkIngestClient{create: create, read: read, close: close, begin: begin}, nil
}

type bulkIngestRoute struct {
	version float64
	jobID   string
	upload  bool
}

func matchBulkIngestRoute(path string) (bulkIngestRoute, bool) {
	parts := strings.Split(strings.Trim(path, "/"), "/")
	if len(parts) < 5 || len(parts) > 7 || parts[0] != "services" || parts[1] != "data" || parts[3] != "jobs" || parts[4] != "ingest" {
		return bulkIngestRoute{}, false
	}
	if !supportedAPIVersion(parts[2]) {
		return bulkIngestRoute{}, false
	}
	version, _ := strconv.ParseFloat(strings.TrimPrefix(parts[2], "v"), 64)
	route := bulkIngestRoute{version: version}
	if len(parts) >= 6 {
		route.jobID = parts[5]
		if route.jobID == "" {
			return bulkIngestRoute{}, false
		}
	}
	if len(parts) == 7 {
		if parts[6] != "batches" {
			return bulkIngestRoute{}, false
		}
		route.upload = true
	}
	return route, true
}

func (s *Service) serveBulkIngest(writer http.ResponseWriter, request *http.Request) bool {
	route, ok := matchBulkIngestRoute(request.URL.Path)
	if !ok {
		return false
	}
	allowed := ""
	switch {
	case route.jobID == "" && request.Method == http.MethodPost:
		s.serveBulkIngestCreate(writer, request, route)
		return true
	case route.jobID == "":
		allowed = http.MethodPost
	case route.upload && request.Method == http.MethodPut:
		s.serveBulkIngestUpload(writer, request, route)
		return true
	case route.upload:
		allowed = http.MethodPut
	case request.Method == http.MethodGet:
		s.serveBulkIngestRead(writer, request, route)
		return true
	case request.Method == http.MethodPatch:
		s.serveBulkIngestClose(writer, request, route)
		return true
	default:
		allowed = http.MethodGet + ", " + http.MethodPatch
	}
	writer.Header().Set("Allow", allowed)
	writeCRMErrorWith(writer, http.StatusMethodNotAllowed, crmwire.SFCodeInvalidRequest, "Method not allowed", writeLegacyJSON)
	return true
}

func (s *Service) beginBulkIngestRequest(writer http.ResponseWriter, request *http.Request) (context.Context, string, context.CancelFunc, bool) {
	s.requests.Add(1)
	if s.draining.Load() {
		s.drainRejected.Add(1)
		s.failures.Add(1)
		writeCRMErrorWith(writer, http.StatusServiceUnavailable, crmwire.SFCodeUnknownException, "Service is draining", writeLegacyJSON)
		return nil, "", nil, false
	}
	bearer, ok := endUserBearer(request.Header.Values("Authorization"))
	if !ok {
		s.localRejected.Add(1)
		s.failures.Add(1)
		writeCRMErrorWith(writer, http.StatusUnauthorized, crmwire.SFCodeInvalidSessionID, "Session expired or invalid", writeLegacyJSON)
		return nil, "", nil, false
	}
	s.inFlight.Add(1)
	ctx, cancel := context.WithTimeout(request.Context(), s.requestTimeout)
	return ctx, bearer, func() { cancel(); s.inFlight.Add(-1) }, true
}

func (s *Service) serveBulkIngestCreate(writer http.ResponseWriter, request *http.Request, route bulkIngestRoute) {
	ctx, bearer, done, ok := s.beginBulkIngestRequest(writer, request)
	if !ok {
		return
	}
	defer done()
	request.Body = http.MaxBytesReader(writer, request.Body, s.maxRequestBytes)
	var body struct {
		Object              string `json:"object"`
		Operation           string `json:"operation"`
		ExternalIDFieldName string `json:"externalIdFieldName,omitempty"`
	}
	decoder := json.NewDecoder(request.Body)
	if err := decoder.Decode(&body); err != nil || requireJSONEOF(decoder) != nil {
		s.rejectBulkIngest(writer, http.StatusBadRequest, crmwire.SFCodeInvalidRequest, "Invalid request body")
		return
	}
	operation, valid := bulkIngestOperation(body.Operation)
	if !valid {
		s.rejectBulkIngest(writer, http.StatusBadRequest, crmwire.SFCodeInvalidOperation, fmt.Sprintf("Invalid operation: '%s'. Valid operations: insert, update, upsert, delete", body.Operation))
		return
	}
	response, err := s.bulkIngest.create(ctx, bearer, bulkingestsemantic.CreateRequest{ObjectID: body.Object, Operation: operation, ExternalIDFieldName: body.ExternalIDFieldName, ClientVersion: route.version})
	if err != nil {
		s.failures.Add(1)
		writeBulkIngestFailure(writer, err, bulkIngestFailureCreate)
		return
	}
	s.successes.Add(1)
	writeLegacyJSON(writer, http.StatusCreated, projectBulkIngestJob(response.Job))
}

func (s *Service) serveBulkIngestRead(writer http.ResponseWriter, request *http.Request, route bulkIngestRoute) {
	ctx, bearer, done, ok := s.beginBulkIngestRequest(writer, request)
	if !ok {
		return
	}
	defer done()
	response, err := s.bulkIngest.read(ctx, bearer, bulkingestsemantic.ReadRequest{JobID: route.jobID})
	if err != nil {
		s.failures.Add(1)
		writeBulkIngestFailure(writer, err, bulkIngestFailureRead)
		return
	}
	s.successes.Add(1)
	writeLegacyJSON(writer, http.StatusOK, projectBulkIngestJob(response.Job))
}

func (s *Service) serveBulkIngestClose(writer http.ResponseWriter, request *http.Request, route bulkIngestRoute) {
	ctx, bearer, done, ok := s.beginBulkIngestRequest(writer, request)
	if !ok {
		return
	}
	defer done()
	request.Body = http.MaxBytesReader(writer, request.Body, s.maxRequestBytes)
	var body struct {
		State string `json:"state"`
	}
	decoder := json.NewDecoder(request.Body)
	if err := decoder.Decode(&body); err != nil || requireJSONEOF(decoder) != nil {
		s.rejectBulkIngest(writer, http.StatusBadRequest, crmwire.SFCodeInvalidRequest, "Invalid request body")
		return
	}
	state, valid := bulkIngestCloseState(body.State)
	if !valid {
		s.rejectBulkIngest(writer, http.StatusBadRequest, crmwire.SFCodeInvalidRequest, "Only UploadComplete or Aborted state changes are allowed")
		return
	}
	response, err := s.bulkIngest.close(ctx, bearer, bulkingestsemantic.CloseRequest{JobID: route.jobID, State: state})
	if err != nil {
		s.failures.Add(1)
		writeBulkIngestFailure(writer, err, bulkIngestFailureClose)
		return
	}
	s.successes.Add(1)
	writeLegacyJSON(writer, http.StatusOK, projectBulkIngestJob(response.Job))
}

func (s *Service) serveBulkIngestUpload(writer http.ResponseWriter, request *http.Request, route bulkIngestRoute) {
	ctx, bearer, done, ok := s.beginBulkIngestRequest(writer, request)
	if !ok {
		return
	}
	defer done()
	if request.ContentLength > int64(bulkingestsemantic.MaxUploadBytes) {
		s.rejectBulkIngest(writer, http.StatusRequestEntityTooLarge, crmwire.SFCodeFileTooLarge, bulkIngestFileTooLargeMessage())
		return
	}
	header := bulkingestsemantic.UploadHeader{JobID: route.jobID}
	if request.ContentLength >= 0 {
		header.ContentLengthKnown = true
		header.ContentLength = uint64(request.ContentLength)
	}
	upload, err := s.bulkIngest.begin(ctx, bearer, header)
	if err != nil {
		s.failures.Add(1)
		writeBulkIngestFailure(writer, err, bulkIngestFailureUpload)
		return
	}
	defer upload.Cancel()
	buffer := make([]byte, bulkingestsemantic.MaxChunkBytes)
	for {
		n, readErr := request.Body.Read(buffer)
		if n > 0 {
			if err := upload.Send(buffer[:n]); err != nil {
				s.failures.Add(1)
				writeBulkIngestFailure(writer, err, bulkIngestFailureUpload)
				return
			}
		}
		if readErr == io.EOF {
			break
		}
		if readErr != nil {
			s.rejectBulkIngest(writer, http.StatusBadRequest, crmwire.SFCodeInvalidRequest, "Invalid request body")
			return
		}
	}
	if _, err := upload.CloseAndRecv(); err != nil {
		s.failures.Add(1)
		writeBulkIngestFailure(writer, err, bulkIngestFailureUpload)
		return
	}
	s.successes.Add(1)
	writer.WriteHeader(http.StatusCreated)
}

func (s *Service) rejectBulkIngest(writer http.ResponseWriter, status int, code, message string) {
	s.localRejected.Add(1)
	s.failures.Add(1)
	writeCRMErrorWith(writer, status, code, message, writeLegacyJSON)
}

func bulkIngestOperation(value string) (bulkingestsemantic.Operation, bool) {
	switch value {
	case "insert":
		return bulkingestsemantic.OperationInsert, true
	case "update":
		return bulkingestsemantic.OperationUpdate, true
	case "upsert":
		return bulkingestsemantic.OperationUpsert, true
	case "delete":
		return bulkingestsemantic.OperationDelete, true
	default:
		return "", false
	}
}

func bulkIngestCloseState(value string) (bulkingestsemantic.State, bool) {
	switch value {
	case "UploadComplete":
		return bulkingestsemantic.StateUploadComplete, true
	case "Aborted":
		return bulkingestsemantic.StateAborted, true
	default:
		return "", false
	}
}

func projectBulkIngestJob(job bulkingestsemantic.Job) any {
	return struct {
		ID                      string  `json:"id"`
		Object                  string  `json:"object"`
		Operation               string  `json:"operation"`
		State                   string  `json:"state"`
		ConcurrencyMode         string  `json:"concurrencyMode"`
		CreatedByID             string  `json:"createdById"`
		CreatedDate             string  `json:"createdDate"`
		SystemModstamp          string  `json:"systemModstamp"`
		ErrorMessage            string  `json:"errorMessage,omitempty"`
		ContentType             string  `json:"contentType"`
		ColumnDelimiter         string  `json:"columnDelimiter"`
		LineEnding              string  `json:"lineEnding"`
		APIVersion              float64 `json:"apiVersion"`
		ExternalIDFieldName     string  `json:"externalIdFieldName,omitempty"`
		NumberRecordsProcessed  uint64  `json:"numberRecordsProcessed"`
		NumberRecordsFailed     uint64  `json:"numberRecordsFailed"`
		TotalProcessingTime     uint64  `json:"totalProcessingTime"`
		APIActiveProcessingTime uint64  `json:"apiActiveProcessingTime"`
		ApexProcessingTime      uint64  `json:"apexProcessingTime"`
	}{
		ID: job.ID, Object: job.ObjectID, Operation: strings.ToLower(string(job.Operation)), State: bulkIngestProductState(job.State),
		ConcurrencyMode: "Parallel", CreatedByID: job.CreatedByID, CreatedDate: job.CreatedAt, SystemModstamp: job.ModifiedAt,
		ErrorMessage: bulkIngestFailureMessage(job.FailureCode), ContentType: "CSV", ColumnDelimiter: "COMMA", LineEnding: "LF", APIVersion: job.ClientVersion,
		ExternalIDFieldName: job.ExternalIDFieldName, NumberRecordsProcessed: job.NumberRecordsProcessed, NumberRecordsFailed: job.NumberRecordsFailed,
		TotalProcessingTime: job.TotalProcessingTime, APIActiveProcessingTime: job.ActiveProcessingTime,
		// NexusCRM has no Apex execution engine. Keep this Salesforce wire field as
		// an explicit L1 compatibility constant instead of teaching the app-blind
		// Runtime contract an Apex-specific fact that no L0 producer can author.
		ApexProcessingTime: 0,
	}
}

func bulkIngestFailureMessage(code bulkingestsemantic.FailureCode) string {
	if code == bulkingestsemantic.FailureProcessingFailed {
		return "Bulk ingest processing failed"
	}
	return ""
}

func bulkIngestProductState(state bulkingestsemantic.State) string {
	switch state {
	case bulkingestsemantic.StateOpen:
		return "Open"
	case bulkingestsemantic.StateUploadComplete:
		return "UploadComplete"
	case bulkingestsemantic.StateInProgress:
		return "InProgress"
	case bulkingestsemantic.StateComplete:
		return "JobComplete"
	case bulkingestsemantic.StateFailed:
		return "Failed"
	case bulkingestsemantic.StateAborted:
		return "Aborted"
	default:
		return ""
	}
}

type bulkIngestFailureOperation uint8

const (
	bulkIngestFailureCreate bulkIngestFailureOperation = iota
	bulkIngestFailureRead
	bulkIngestFailureClose
	bulkIngestFailureUpload
)

func writeBulkIngestFailure(writer http.ResponseWriter, err error, operation bulkIngestFailureOperation) {
	var failure *commonsemantic.Failure
	if !errors.As(err, &failure) {
		writeRuntimeFailureWith(writer, err, writeLegacyJSON)
		return
	}
	switch failure.Category {
	case commonsemantic.FailureInvalidRequest:
		code := crmwire.SFCodeInvalidRequest
		message := "Invalid request"
		if failure.InvalidComponent == commonsemantic.InvalidComponentObjectID {
			code, message = crmwire.SFCodeInvalidObject, "Invalid object"
		} else if failure.InvalidComponent == commonsemantic.InvalidComponentOperation {
			if operation == bulkIngestFailureCreate {
				code, message = crmwire.SFCodeInvalidOperation, "Invalid operation"
			} else {
				code, message = crmwire.SFCodeInvalidJobState, "Job cannot transition to the requested state"
			}
		}
		writeCRMErrorWith(writer, http.StatusBadRequest, code, message, writeLegacyJSON)
	case commonsemantic.FailureResourceExhausted:
		if failure.LimitKind == commonsemantic.LimitUploadBytes {
			writeCRMErrorWith(writer, http.StatusRequestEntityTooLarge, crmwire.SFCodeFileTooLarge, bulkIngestFileTooLargeMessage(), writeLegacyJSON)
			return
		}
		writeRuntimeFailureWith(writer, err, writeLegacyJSON)
	case commonsemantic.FailureForbidden:
		code := crmwire.SFCodeForbidden
		if operation == bulkIngestFailureCreate {
			code = crmwire.SFCodeInsufficientPerms
		}
		writeCRMErrorWith(writer, http.StatusForbidden, code, "Insufficient access", writeLegacyJSON)
	case commonsemantic.FailureMissing:
		writeCRMErrorWith(writer, http.StatusNotFound, crmwire.SFCodeNotFound, "Job not found", writeLegacyJSON)
	default:
		writeRuntimeFailureWith(writer, err, writeLegacyJSON)
	}
}

func bulkIngestFileTooLargeMessage() string {
	return fmt.Sprintf("File size exceeds maximum allowed size of %d MB", bulkingestsemantic.MaxUploadBytes/(1024*1024))
}