package dialectsfcrm

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

	soql "github.com/nexuscrm/dialect-sf-crm-soql"
	"github.com/nexuscrm/dialect-sf-crm/wire"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/dialect"
	sharederrors "github.com/nexuscrm/shared/pkg/errors"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/shared/pkg/queryspec"
)

const bulkQueryJobNotCompleteCode = "JOB_NOT_COMPLETE"

type bulkQueryCreateRequest struct {
	Operation string `json:"operation"`
	Query     string `json:"query"`
}

type bulkQueryHandler struct {
	jobs   contract.BulkJobs
	query  contract.Query
	policy contract.Policy
}

func CreateBulkQueryJobNeutralHandler(ctx context.Context, req *dialect.Request, deps dialect.Deps) (*dialect.Response, error) {
	return bulkQueryHandlerFromDeps(deps).createQueryJob(ctx, req)
}

func GetBulkQueryJobNeutralHandler(ctx context.Context, req *dialect.Request, deps dialect.Deps) (*dialect.Response, error) {
	return bulkQueryHandlerFromDeps(deps).getQueryJob(ctx, req)
}

func GetBulkQueryResultsNeutralHandler(ctx context.Context, req *dialect.Request, deps dialect.Deps) (*dialect.Response, error) {
	return bulkQueryHandlerFromDeps(deps).getQueryResults(ctx, req)
}

func bulkQueryHandlerFromDeps(deps dialect.Deps) bulkQueryHandler {
	return bulkQueryHandler{
		jobs:   deps.Contracts.BulkJobs,
		query:  deps.Contracts.Query,
		policy: deps.Contracts.Policy,
	}
}

func (h bulkQueryHandler) createQueryJob(ctx context.Context, req *dialect.Request) (*dialect.Response, error) {
	if req == nil || req.User == nil {
		return &dialect.Response{Status: http.StatusUnauthorized, Body: wire.SFErrorResponse(wire.SFCodeUnauthorized, "Authentication required")}, nil
	}
	if h.jobs == nil || h.query == nil {
		return bulkQueryInternalErrorResponse("create query job contracts unavailable", nil), nil
	}

	var body bulkQueryCreateRequest
	if err := json.Unmarshal(req.Body, &body); err != nil {
		return &dialect.Response{Status: http.StatusBadRequest, Body: wire.SFErrorResponse(wire.SFCodeInvalidRequest, err.Error())}, nil
	}
	if body.Query == "" {
		return &dialect.Response{Status: http.StatusBadRequest, Body: wire.SFErrorResponse(wire.SFCodeMissingQuery, "Query is required")}, nil
	}
	if body.Operation == "" {
		body.Operation = "query"
	}
	operation, resp := normalizeBulkQueryOperation(body.Operation)
	if resp != nil {
		return resp, nil
	}

	now := time.Now()
	job := &models.BulkJob{
		ID:              createBulkQueryJobID(now),
		Operation:       operation,
		Object:          "",
		State:           models.JobStateUploadComplete,
		JobType:         "V2Query",
		ConcurrencyMode: models.ConcurrencyParallel,
		ContentType:     "CSV",
		ColumnDelimiter: "COMMA",
		LineEnding:      "LF",
		ApiVersion:      bulkAPIVersionFloat(req.Version),
		CreatedById:     req.User.ID,
		CreatedDate:     now,
		SystemModstamp:  now,
	}

	if err := h.jobs.CreateJob(ctx, job); err != nil {
		return bulkQueryInternalErrorResponse("create query job failed", err), nil
	}

	job.ContentUrl = body.Query
	if err := h.jobs.UpdateJob(ctx, job); err != nil {
		return bulkQueryInternalErrorResponse("store query on bulk query job failed", err), nil
	}

	go processBulkQueryJob(job.ID, body.Query, req.User, h.jobs, h.query)

	return &dialect.Response{Status: http.StatusCreated, Body: job}, nil
}

func (h bulkQueryHandler) getQueryJob(ctx context.Context, req *dialect.Request) (*dialect.Response, error) {
	if req == nil || req.User == nil {
		return &dialect.Response{Status: http.StatusUnauthorized, Body: wire.SFErrorResponse(wire.SFCodeUnauthorized, "Authentication required")}, nil
	}
	if h.jobs == nil {
		return bulkQueryInternalErrorResponse("get query job contract unavailable", nil), nil
	}

	job, resp := h.getQueryJobOrError(ctx, jobIDParam(req))
	if resp != nil {
		return resp, nil
	}
	if !h.canAccessJob(ctx, req.User, job) {
		return &dialect.Response{Status: http.StatusForbidden, Body: wire.SFErrorResponse(wire.SFCodeForbidden, "Insufficient access to this job")}, nil
	}

	return &dialect.Response{Status: http.StatusOK, Body: bulkQueryJobStatusBody(job)}, nil
}

func (h bulkQueryHandler) getQueryResults(ctx context.Context, req *dialect.Request) (*dialect.Response, error) {
	if req == nil || req.User == nil {
		return &dialect.Response{Status: http.StatusUnauthorized, Body: wire.SFErrorResponse(wire.SFCodeUnauthorized, "Authentication required")}, nil
	}
	if h.jobs == nil || h.query == nil {
		return bulkQueryInternalErrorResponse("get query results contracts unavailable", nil), nil
	}

	job, resp := h.getQueryJobOrError(ctx, jobIDParam(req))
	if resp != nil {
		return resp, nil
	}
	if !h.canAccessJob(ctx, req.User, job) {
		return &dialect.Response{Status: http.StatusForbidden, Body: wire.SFErrorResponse(wire.SFCodeForbidden, "Insufficient access to this job")}, nil
	}
	if job.State != models.JobStateJobComplete {
		return &dialect.Response{
			Status: http.StatusBadRequest,
			Body:   wire.SFErrorResponse(bulkQueryJobNotCompleteCode, fmt.Sprintf("Job is in state %s, results not available", job.State)),
		}, nil
	}
	if job.ContentUrl == "" {
		return &dialect.Response{Status: http.StatusBadRequest, Body: wire.SFErrorResponse(wire.SFCodeNoQuery, "No query stored for job")}, nil
	}

	limit := optionalPositiveInt(queryValue(req, "maxRecords"))
	offset := optionalNonNegativeInt(queryValue(req, "locator"))
	rows, err := executeBulkQuery(ctx, h.query, job.ContentUrl, req.User, limit, offset, bulkQueryScanAll(job))
	if err != nil {
		return bulkQueryExecutionErrorResponse(req, err), nil
	}

	return &dialect.Response{
		Status: http.StatusOK,
		Headers: map[string]string{
			"Sforce-Locator":         "null",
			"Sforce-NumberOfRecords": fmt.Sprintf("%d", len(rows)),
		},
		Body: rows,
	}, nil
}

func (h bulkQueryHandler) getQueryJobOrError(ctx context.Context, jobID string) (*models.BulkJob, *dialect.Response) {
	job, err := h.jobs.GetJob(ctx, jobID)
	if err != nil {
		if sharederrors.IsNotFound(err) {
			return nil, bulkQueryJobNotFoundResponse()
		}
		return nil, bulkQueryInternalErrorResponse("get bulk query job failed", err)
	}
	if job == nil {
		return nil, bulkQueryJobNotFoundResponse()
	}
	return job, nil
}

func (h bulkQueryHandler) canAccessJob(ctx context.Context, user *models.UserSession, job *models.BulkJob) bool {
	if user == nil || job == nil {
		return false
	}
	if job.CreatedById == user.ID {
		return true
	}
	capabilityPolicy, ok := h.policy.(contract.CapabilityPolicy)
	if !ok {
		return false
	}
	return capabilityPolicy.HasCapability(ctx, user, constants.SuperUserCapability)
}

func processBulkQueryJob(jobID, rawQuery string, user *models.UserSession, jobs contract.BulkJobs, query contract.Query) {
	if jobs == nil || query == nil {
		return
	}
	ctx := context.Background()
	job, err := jobs.GetJob(ctx, jobID)
	if err != nil || job == nil {
		return
	}

	parsedReq, err := soql.ParseSOQL(rawQuery)
	if err != nil {
		job.State = models.JobStateFailed
		job.ErrorMessage = fmt.Sprintf("Invalid SOQL: %s", err.Error())
		job.SystemModstamp = time.Now()
		_ = jobs.UpdateJob(ctx, job)
		return
	}

	result, err := query.Query(ctx, bulkQuerySpecFromParsedSOQL(parsedReq, parsedReq.Limit, parsedReq.Offset, bulkQueryScanAll(job)), user)
	if err != nil {
		job.State = models.JobStateFailed
		job.ErrorMessage = fmt.Sprintf("Query execution failed: %s", err.Error())
		job.SystemModstamp = time.Now()
		_ = jobs.UpdateJob(ctx, job)
		return
	}

	job.State = models.JobStateJobComplete
	job.Object = parsedReq.ObjectAPIName
	if result != nil {
		job.NumberRecordsProcessed = len(result.Records)
	}
	job.SystemModstamp = time.Now()
	_ = jobs.UpdateJob(ctx, job)
}

func executeBulkQuery(ctx context.Context, query contract.Query, rawQuery string, user *models.UserSession, overrideLimit *int, overrideOffset *int, scanAll bool) ([]map[string]interface{}, error) {
	if query == nil {
		return nil, fmt.Errorf("query service unavailable")
	}
	if !strings.HasPrefix(strings.TrimSpace(strings.ToUpper(rawQuery)), "SELECT") {
		return nil, NewValidationError("query", "invalid SOQL: must start with SELECT")
	}
	parsedReq, err := soql.ParseSOQL(rawQuery)
	if err != nil {
		return nil, NewValidationError("query", fmt.Sprintf("invalid SOQL: %v", err))
	}

	limit := parsedReq.Limit
	if overrideLimit != nil {
		limit = *overrideLimit
	}
	offset := parsedReq.Offset
	if overrideOffset != nil {
		offset = *overrideOffset
	}

	result, err := query.Query(ctx, bulkQuerySpecFromParsedSOQL(parsedReq, limit, offset, scanAll), user)
	if err != nil {
		return nil, err
	}
	if result == nil {
		return []map[string]interface{}{}, nil
	}

	records := make([]map[string]interface{}, len(result.Records))
	for i, rec := range result.Records {
		records[i] = map[string]interface{}(projectRecordToV63(rec, v63RecordProjectionFieldsOnly))
	}
	return records, nil
}

// bulkQuerySpecFromParsedSOQL keeps the Bulk API's own slim projection and paging
// overrides (limit/offset come from the bulk pager, not the SOQL), but forwards
// the full multi-column ORDER BY through the shared models.SortCriteriaToQuerySpec
// converter so /jobs/query preserves ORDER BY exactly like /query and /tooling/query.
func bulkQuerySpecFromParsedSOQL(req *models.QueryRequest, limit int, offset int, scanAll bool) queryspec.QuerySpec {
	if req == nil {
		return queryspec.QuerySpec{}
	}
	return queryspec.QuerySpec{
		ObjectAPIName: req.ObjectAPIName,
		Filter:        req.Filter,
		SortField:     req.SortField,
		SortDirection: req.SortDirection,
		OrderBy:       models.SortCriteriaToQuerySpec(req.OrderBy),
		Limit:         limit,
		Offset:        offset,
		ScanAll:       scanAll,
	}
}

func normalizeBulkQueryOperation(raw string) (models.BulkOperation, *dialect.Response) {
	switch strings.TrimSpace(raw) {
	case "", string(models.OperationQuery):
		return models.OperationQuery, nil
	case string(models.OperationQueryAll):
		return models.OperationQueryAll, nil
	default:
		return "", &dialect.Response{
			Status: http.StatusBadRequest,
			Body:   wire.SFErrorResponse(wire.SFCodeInvalidRequest, fmt.Sprintf("Unsupported bulk query operation %q", raw)),
		}
	}
}

func bulkQueryScanAll(job *models.BulkJob) bool {
	return job != nil && job.Operation == models.OperationQueryAll
}

func bulkQueryExecutionErrorResponse(req *dialect.Request, err error) *dialect.Response {
	if err == nil {
		return nil
	}
	if sharederrors.IsInternal(err) {
		return bulkQueryInternalErrorResponse("execute bulk query failed", err)
	}
	if resp := respondServiceError(req, err); resp != nil {
		return resp
	}
	if _, message, ok := sharederrors.ValidationDetails(err); ok {
		return &dialect.Response{Status: http.StatusBadRequest, Body: wire.SFErrorResponse(wire.SFCodeQueryFailed, message)}
	}
	var statusErr httpStatusError
	if errors.As(err, &statusErr) && statusErr.HTTPStatus() >= http.StatusBadRequest && statusErr.HTTPStatus() < http.StatusInternalServerError {
		return &dialect.Response{Status: http.StatusBadRequest, Body: wire.SFErrorResponse(wire.SFCodeQueryFailed, err.Error())}
	}
	return bulkQueryInternalErrorResponse("execute bulk query failed", err)
}

func bulkQueryJobStatusBody(job *models.BulkJob) map[string]interface{} {
	return map[string]interface{}{
		"id":                     job.ID,
		"operation":              job.Operation,
		"object":                 job.Object,
		"createdById":            job.CreatedById,
		"createdDate":            job.CreatedDate.Format(time.RFC3339),
		"systemModstamp":         job.SystemModstamp.Format(time.RFC3339),
		"state":                  job.State,
		"jobType":                job.JobType,
		"numberRecordsProcessed": 0,
	}
}

func createBulkQueryJobID(now time.Time) string {
	return fmt.Sprintf("750%s%04d", now.Format("20060102150405"), now.Nanosecond()/100000)
}

func bulkAPIVersionFloat(version string) float64 {
	raw := strings.TrimPrefix(apiVersionOrDefault(version), "v")
	parsed, err := strconv.ParseFloat(raw, 64)
	if err != nil {
		return 63.0
	}
	return parsed
}

func jobIDParam(req *dialect.Request) string {
	if req == nil || req.Params == nil {
		return ""
	}
	return req.Params["jobId"]
}

func optionalPositiveInt(raw string) *int {
	n, err := strconv.Atoi(raw)
	if err != nil || n <= 0 {
		return nil
	}
	return &n
}

func optionalNonNegativeInt(raw string) *int {
	n, err := strconv.Atoi(raw)
	if err != nil || n < 0 {
		return nil
	}
	return &n
}

func bulkQueryJobNotFoundResponse() *dialect.Response {
	return &dialect.Response{Status: http.StatusNotFound, Body: wire.SFErrorResponse(wire.SFCodeNotFound, "Job not found")}
}

func bulkQueryInternalErrorResponse(context string, err error) *dialect.Response {
	if err != nil {
		log.Printf("bulk query: %s: %v", context, err)
	} else {
		log.Printf("bulk query: %s", context)
	}
	return &dialect.Response{
		Status: http.StatusInternalServerError,
		Body:   wire.SFErrorResponse(wire.SFCodeInternalError, "An internal server error occurred. Contact your system administrator."),
	}
}