package connector_delivery

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"io"
	"strings"
	"time"

	connector_oauth "github.com/pingcap-inc/nexuscrm-connectors/features/connector/oauth"
	"github.com/pingcap-inc/nexuscrm-connectors/internal/sqltx"
)

var errTransactionCleanup = errors.New("connected delivery transaction cleanup failed")

type sqlRepository struct {
	db *sql.DB
}

func newSQLRepository(db *sql.DB) *sqlRepository {
	return &sqlRepository{db: db}
}

func (r *sqlRepository) Lookup(ctx context.Context, requestID, userID string) (Job, error) {
	if r == nil || r.db == nil {
		return Job{}, ErrDependencyMissing
	}
	job, err := scanJob(r.db.QueryRowContext(ctx, jobSelect+`
 WHERE JobId = ? AND OriginatingUserId = ? LIMIT 1`, requestID, userID))
	if errors.Is(err, sql.ErrNoRows) {
		return Job{}, ErrDeliveryNotFound
	}
	if err != nil {
		return Job{}, err
	}
	return *job, nil
}

func (r *sqlRepository) Reserve(ctx context.Context, request Request, principal connector_oauth.ConnectedPrincipal, payload payloadRecord, now time.Time) (job Job, err error) {
	if r == nil || r.db == nil {
		return Job{}, ErrDependencyMissing
	}
	err = r.withTx(ctx, func(tx *sql.Tx) error {
		if err := lockUser(ctx, tx, request.UserID); err != nil {
			return err
		}
		existing, err := loadJob(ctx, tx, request.RequestID, true)
		if err == nil {
			if err := verifyExistingRequest(ctx, tx, *existing, request, principal, payload); err != nil {
				return err
			}
			job = *existing
			return nil
		}
		if !errors.Is(err, sql.ErrNoRows) {
			return err
		}
		if principal.Generation != request.ExpectedGeneration {
			return connector_oauth.ErrGenerationMismatch
		}
		if err := lockConnectedPrincipal(ctx, tx, principal, request); err != nil {
			return err
		}
		if _, err := tx.ExecContext(ctx, `
INSERT INTO _System_ConnectedDeliveryPayload
       (JobId, PayloadBase64, PayloadSha256, EncodedSize, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, ?)`, request.RequestID, payload.Base64, payload.SHA256,
			payload.EncodedSize, now, now); err != nil {
			return classifyConflict(err)
		}
		if _, err := tx.ExecContext(ctx, `
INSERT INTO _System_ConnectedDeliveryJob
       (JobId, ConnectedPrincipalId, PrincipalGeneration, OriginatingUserId, State,
        PayloadSha256, PayloadEncodedSize, SourceObjectApiName, SourceRecordId,
        CompletionSnapshot, ClaimToken, LeaseExpiresAt, Fence, DispatchStartedAt,
        AcceptedAt, RemoteReceiptId, RemoteConversationId, ResponseStatus,
        ResultEvidence, FailureCode, LocalAttemptCount, NextLocalAttemptAt,
        CompletedAt, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, NULL, 0, NULL, NULL, NULL, NULL,
        NULL, NULL, NULL, 0, NULL, NULL, ?, ?)`,
			request.RequestID, principal.ID, request.ExpectedGeneration, request.UserID,
			StateReserved, payload.SHA256, payload.EncodedSize, request.SourceObject,
			request.SourceRecordID, []byte(request.CompletionSnapshot), now, now); err != nil {
			return classifyConflict(err)
		}
		job = Job{
			RequestID: request.RequestID, ConnectedPrincipalID: principal.ID,
			PrincipalGeneration: request.ExpectedGeneration, OriginatingUserID: request.UserID,
			State: StateReserved, PayloadSHA256: payload.SHA256,
			PayloadEncodedSize: payload.EncodedSize, SourceObject: request.SourceObject,
			SourceRecordID:     request.SourceRecordID,
			CompletionSnapshot: append(json.RawMessage(nil), request.CompletionSnapshot...),
			CreatedDate:        now, LastModifiedDate: now,
		}
		return nil
	})
	return job, err
}

func (r *sqlRepository) BeginDispatch(ctx context.Context, requestID, userID, claimToken string, now, leaseUntil time.Time) (job Job, claim *dispatchClaim, err error) {
	if r == nil || r.db == nil {
		return Job{}, nil, ErrDependencyMissing
	}
	now, leaseUntil = schemaTime(now), schemaTime(leaseUntil)
	if strings.TrimSpace(claimToken) == "" || !leaseUntil.After(now) {
		return Job{}, nil, ErrInvalidInput
	}
	err = r.withTx(ctx, func(tx *sql.Tx) error {
		if err := lockUser(ctx, tx, userID); err != nil {
			return err
		}
		loaded, err := loadJobForUser(ctx, tx, requestID, userID, true)
		if errors.Is(err, sql.ErrNoRows) {
			return ErrRequestConflict
		}
		if err != nil {
			return err
		}
		job = *loaded
		switch job.State {
		case StateDispatching:
			if job.LeaseExpiresAt != nil && job.LeaseExpiresAt.After(now) {
				return ErrDispatchInProgress
			}
			if err := updateExpiredDispatch(ctx, tx, &job, now); err != nil {
				return err
			}
			return nil
		case StateAccepted, StateCompleted, StateFailed, StateUncertain:
			return nil
		case StateReserved:
		default:
			return ErrRequestConflict
		}
		connected, err := principalGenerationConnected(ctx, tx, job)
		if err != nil {
			return err
		}
		if !connected {
			return failReservedGeneration(ctx, tx, &job, now)
		}
		result, err := tx.ExecContext(ctx, `
UPDATE _System_ConnectedDeliveryJob
   SET State = ?, ClaimToken = ?, LeaseExpiresAt = ?, Fence = Fence + 1,
       DispatchStartedAt = ?, LastModifiedDate = ?
 WHERE JobId = ? AND OriginatingUserId = ? AND State = ?
   AND ClaimToken IS NULL AND LeaseExpiresAt IS NULL AND DispatchStartedAt IS NULL
   AND AcceptedAt IS NULL AND RemoteReceiptId IS NULL AND RemoteConversationId IS NULL
   AND ResponseStatus IS NULL AND ResultEvidence IS NULL AND FailureCode IS NULL
   AND LocalAttemptCount = 0 AND NextLocalAttemptAt IS NULL AND CompletedAt IS NULL`,
			StateDispatching, claimToken, leaseUntil, now, now, job.RequestID,
			userID, StateReserved)
		if err != nil {
			return err
		}
		won, err := exactlyOne(result)
		if err != nil {
			return err
		}
		if !won {
			return ErrOwnershipLost
		}
		job.State = StateDispatching
		job.ClaimToken = claimToken
		job.LeaseExpiresAt = &leaseUntil
		job.DispatchStartedAt = &now
		job.Fence++
		job.LastModifiedDate = now
		claim = &dispatchClaim{Job: job, ClaimToken: claimToken, Fence: job.Fence}
		return nil
	})
	return job, claim, err
}

func (r *sqlRepository) ReleaseDispatch(ctx context.Context, claim dispatchClaim, now time.Time) (job Job, err error) {
	if r == nil || r.db == nil {
		return Job{}, ErrDependencyMissing
	}
	now = schemaTime(now)
	err = r.withTx(ctx, func(tx *sql.Tx) error {
		if err := lockUser(ctx, tx, claim.Job.OriginatingUserID); err != nil {
			return err
		}
		current, err := loadJobForUser(ctx, tx, claim.Job.RequestID, claim.Job.OriginatingUserID, true)
		if err != nil {
			return err
		}
		if current.State != StateDispatching || current.ClaimToken != claim.ClaimToken ||
			current.Fence != claim.Fence || current.LeaseExpiresAt == nil ||
			claim.Job.LeaseExpiresAt == nil || !current.LeaseExpiresAt.Equal(*claim.Job.LeaseExpiresAt) ||
			current.DispatchStartedAt == nil || claim.Job.DispatchStartedAt == nil ||
			!current.DispatchStartedAt.Equal(*claim.Job.DispatchStartedAt) {
			return ErrOwnershipLost
		}
		result, err := tx.ExecContext(ctx, `
UPDATE _System_ConnectedDeliveryJob
   SET State = ?, ClaimToken = NULL, LeaseExpiresAt = NULL,
       DispatchStartedAt = NULL, LastModifiedDate = ?
 WHERE JobId = ? AND OriginatingUserId = ? AND State = ?
   AND BINARY ClaimToken = BINARY ? AND Fence = ? AND LeaseExpiresAt = ?
   AND DispatchStartedAt = ? AND AcceptedAt IS NULL AND RemoteReceiptId IS NULL
   AND RemoteConversationId IS NULL AND ResponseStatus IS NULL
   AND ResultEvidence IS NULL AND FailureCode IS NULL AND LocalAttemptCount = 0
   AND NextLocalAttemptAt IS NULL AND CompletedAt IS NULL`,
			StateReserved, now, claim.Job.RequestID, claim.Job.OriginatingUserID,
			StateDispatching, claim.ClaimToken, claim.Fence, current.LeaseExpiresAt,
			current.DispatchStartedAt)
		if err != nil {
			return err
		}
		won, err := exactlyOne(result)
		if err != nil {
			return err
		}
		if !won {
			return ErrOwnershipLost
		}
		loaded, err := loadJob(ctx, tx, claim.Job.RequestID, false)
		if err == nil {
			job = *loaded
		}
		return err
	})
	return job, err
}

func (r *sqlRepository) FinishDispatch(ctx context.Context, claim dispatchClaim, terminal terminalDispatch, now time.Time) (job Job, err error) {
	if r == nil || r.db == nil {
		return Job{}, ErrDependencyMissing
	}
	if terminal.State != StateAccepted && terminal.State != StateFailed && terminal.State != StateUncertain {
		return Job{}, ErrInvalidInput
	}
	if err := validateTerminalDispatch(terminal); err != nil {
		return Job{}, err
	}
	err = r.withTx(ctx, func(tx *sql.Tx) error {
		if err := lockUser(ctx, tx, claim.Job.OriginatingUserID); err != nil {
			return err
		}
		current, err := loadJobForUser(ctx, tx, claim.Job.RequestID, claim.Job.OriginatingUserID, true)
		if err != nil {
			return err
		}
		if current.State != StateDispatching || current.ClaimToken != claim.ClaimToken ||
			current.Fence != claim.Fence || current.LeaseExpiresAt == nil ||
			claim.Job.LeaseExpiresAt == nil || !current.LeaseExpiresAt.Equal(*claim.Job.LeaseExpiresAt) ||
			current.DispatchStartedAt == nil || claim.Job.DispatchStartedAt == nil ||
			!current.DispatchStartedAt.Equal(*claim.Job.DispatchStartedAt) {
			return ErrOwnershipLost
		}
		acceptedAt, completedAt := any(nil), any(nil)
		if terminal.State == StateAccepted {
			acceptedAt = now
		} else {
			completedAt = now
		}
		result, err := tx.ExecContext(ctx, `
UPDATE _System_ConnectedDeliveryJob
   SET State = ?, AcceptedAt = ?, RemoteReceiptId = ?, RemoteConversationId = ?,
       ResponseStatus = ?, FailureCode = ?, ClaimToken = NULL, LeaseExpiresAt = NULL,
       CompletedAt = ?, LastModifiedDate = ?
 WHERE JobId = ? AND OriginatingUserId = ? AND State = ?
   AND BINARY ClaimToken = BINARY ? AND Fence = ? AND LeaseExpiresAt = ?
   AND DispatchStartedAt = ? AND AcceptedAt IS NULL AND RemoteReceiptId IS NULL
   AND RemoteConversationId IS NULL AND ResponseStatus IS NULL
   AND ResultEvidence IS NULL AND FailureCode IS NULL AND LocalAttemptCount = 0
   AND NextLocalAttemptAt IS NULL AND CompletedAt IS NULL`,
			terminal.State, acceptedAt, nullableString(terminal.RemoteReceiptID),
			nullableString(terminal.RemoteConversationID), nullableInt(terminal.ResponseStatus),
			nullableString(terminal.FailureCode), completedAt, now, claim.Job.RequestID,
			claim.Job.OriginatingUserID, StateDispatching, claim.ClaimToken, claim.Fence,
			current.LeaseExpiresAt, current.DispatchStartedAt)
		if err != nil {
			return err
		}
		won, err := exactlyOne(result)
		if err != nil {
			return err
		}
		if !won {
			return ErrOwnershipLost
		}
		loaded, err := loadJob(ctx, tx, claim.Job.RequestID, false)
		if err == nil {
			job = *loaded
		}
		return err
	})
	return job, err
}

func (r *sqlRepository) Complete(ctx context.Context, requestID, userID string, now time.Time) (job Job, err error) {
	if r == nil || r.db == nil {
		return Job{}, ErrDependencyMissing
	}
	err = r.withTx(ctx, func(tx *sql.Tx) error {
		if err := lockUser(ctx, tx, userID); err != nil {
			return err
		}
		loaded, err := loadJobForUser(ctx, tx, requestID, userID, true)
		if errors.Is(err, sql.ErrNoRows) {
			return ErrRequestConflict
		}
		if err != nil {
			return err
		}
		job = *loaded
		if job.State == StateCompleted {
			return nil
		}
		if job.State != StateAccepted {
			return ErrCompletionInvalid
		}
		result, err := tx.ExecContext(ctx, `
UPDATE _System_ConnectedDeliveryJob
   SET State = ?, CompletedAt = ?, LastModifiedDate = ?
 WHERE JobId = ? AND OriginatingUserId = ? AND State = ?`,
			StateCompleted, now, now, requestID, userID, StateAccepted)
		if err != nil {
			return err
		}
		won, err := exactlyOne(result)
		if err != nil {
			return err
		}
		if !won {
			return ErrOwnershipLost
		}
		job.State = StateCompleted
		job.CompletedAt = &now
		job.LastModifiedDate = now
		return nil
	})
	return job, err
}

func verifyExistingRequest(ctx context.Context, tx *sql.Tx, job Job, request Request, principal connector_oauth.ConnectedPrincipal, payload payloadRecord) error {
	if job.OriginatingUserID != request.UserID || job.ConnectedPrincipalID != principal.ID ||
		job.PrincipalGeneration != request.ExpectedGeneration ||
		job.PayloadSHA256 != payload.SHA256 || job.PayloadEncodedSize != payload.EncodedSize ||
		job.SourceObject != request.SourceObject || job.SourceRecordID != request.SourceRecordID ||
		!sameJSON(job.CompletionSnapshot, request.CompletionSnapshot) {
		return ErrRequestConflict
	}
	var storedBase64, storedSHA string
	var storedSize int64
	if err := tx.QueryRowContext(ctx, `
SELECT PayloadBase64, PayloadSha256, EncodedSize
  FROM _System_ConnectedDeliveryPayload
 WHERE JobId = ? LIMIT 1 FOR UPDATE`, request.RequestID).
		Scan(&storedBase64, &storedSHA, &storedSize); err != nil {
		return ErrRequestConflict
	}
	if storedBase64 != payload.Base64 || storedSHA != payload.SHA256 || storedSize != payload.EncodedSize {
		return ErrRequestConflict
	}
	return nil
}

func sameJSON(left, right json.RawMessage) bool {
	if !validJSONWithUniqueObjectKeys(left) || !validJSONWithUniqueObjectKeys(right) {
		return false
	}
	if bytes.Equal(left, right) {
		return true
	}
	var leftValue, rightValue any
	leftDecoder := json.NewDecoder(bytes.NewReader(left))
	leftDecoder.UseNumber()
	rightDecoder := json.NewDecoder(bytes.NewReader(right))
	rightDecoder.UseNumber()
	if leftDecoder.Decode(&leftValue) != nil || rightDecoder.Decode(&rightValue) != nil {
		return false
	}
	leftCanonical, leftErr := json.Marshal(leftValue)
	rightCanonical, rightErr := json.Marshal(rightValue)
	return leftErr == nil && rightErr == nil && bytes.Equal(leftCanonical, rightCanonical)
}

func validJSONWithUniqueObjectKeys(value json.RawMessage) bool {
	decoder := json.NewDecoder(bytes.NewReader(value))
	decoder.UseNumber()
	if !consumeUniqueJSONValue(decoder) {
		return false
	}
	_, err := decoder.Token()
	return errors.Is(err, io.EOF)
}

func consumeUniqueJSONValue(decoder *json.Decoder) bool {
	token, err := decoder.Token()
	if err != nil {
		return false
	}
	delimiter, ok := token.(json.Delim)
	if !ok {
		return true
	}
	switch delimiter {
	case '{':
		keys := make(map[string]struct{})
		for decoder.More() {
			keyToken, err := decoder.Token()
			key, ok := keyToken.(string)
			if err != nil || !ok {
				return false
			}
			if _, duplicate := keys[key]; duplicate {
				return false
			}
			keys[key] = struct{}{}
			if !consumeUniqueJSONValue(decoder) {
				return false
			}
		}
		closing, err := decoder.Token()
		return err == nil && closing == json.Delim('}')
	case '[':
		for decoder.More() {
			if !consumeUniqueJSONValue(decoder) {
				return false
			}
		}
		closing, err := decoder.Token()
		return err == nil && closing == json.Delim(']')
	default:
		return false
	}
}

func lockConnectedPrincipal(ctx context.Context, tx *sql.Tx, principal connector_oauth.ConnectedPrincipal, request Request) error {
	var id string
	err := tx.QueryRowContext(ctx, `
SELECT Id FROM _System_ConnectedPrincipal
 WHERE Id = ? AND UserId = ? AND ConnectorKey = ? AND Generation = ? AND State = ?
 LIMIT 1 FOR UPDATE`, principal.ID, request.UserID, request.ConnectorKey,
		request.ExpectedGeneration, connector_oauth.PrincipalConnected).Scan(&id)
	if errors.Is(err, sql.ErrNoRows) {
		return connector_oauth.ErrGenerationMismatch
	}
	return err
}

func principalGenerationConnected(ctx context.Context, tx *sql.Tx, job Job) (bool, error) {
	var id string
	err := tx.QueryRowContext(ctx, `
SELECT Id FROM _System_ConnectedPrincipal
 WHERE Id = ? AND UserId = ? AND Generation = ? AND State = ?
 LIMIT 1 FOR UPDATE`, job.ConnectedPrincipalID, job.OriginatingUserID,
		job.PrincipalGeneration, connector_oauth.PrincipalConnected).Scan(&id)
	if errors.Is(err, sql.ErrNoRows) {
		return false, nil
	}
	return err == nil, err
}

func failReservedGeneration(ctx context.Context, tx *sql.Tx, job *Job, now time.Time) error {
	result, err := tx.ExecContext(ctx, `
UPDATE _System_ConnectedDeliveryJob
   SET State = ?, FailureCode = ?, CompletedAt = ?, LastModifiedDate = ?
 WHERE JobId = ? AND OriginatingUserId = ? AND State = ? AND Fence = ?
   AND ClaimToken IS NULL AND LeaseExpiresAt IS NULL AND DispatchStartedAt IS NULL
   AND AcceptedAt IS NULL AND RemoteReceiptId IS NULL AND RemoteConversationId IS NULL
   AND ResponseStatus IS NULL AND ResultEvidence IS NULL AND FailureCode IS NULL
   AND LocalAttemptCount = 0 AND NextLocalAttemptAt IS NULL AND CompletedAt IS NULL`,
		StateFailed, "connection_generation_changed", now, now, job.RequestID,
		job.OriginatingUserID, StateReserved, job.Fence)
	if err != nil {
		return err
	}
	won, err := exactlyOne(result)
	if err != nil {
		return err
	}
	if !won {
		return ErrOwnershipLost
	}
	job.State = StateFailed
	job.FailureCode = "connection_generation_changed"
	job.CompletedAt = &now
	job.LastModifiedDate = now
	return nil
}

func updateExpiredDispatch(ctx context.Context, tx *sql.Tx, job *Job, now time.Time) error {
	result, err := tx.ExecContext(ctx, `
UPDATE _System_ConnectedDeliveryJob
   SET State = ?, FailureCode = ?, ClaimToken = NULL, LeaseExpiresAt = NULL,
       CompletedAt = ?, LastModifiedDate = ?
 WHERE JobId = ? AND OriginatingUserId = ? AND State = ? AND Fence = ?
   AND BINARY ClaimToken = BINARY ? AND LeaseExpiresAt = ? AND LeaseExpiresAt <= ?
   AND DispatchStartedAt = ? AND AcceptedAt IS NULL AND RemoteReceiptId IS NULL
   AND RemoteConversationId IS NULL AND ResponseStatus IS NULL
   AND ResultEvidence IS NULL AND FailureCode IS NULL AND LocalAttemptCount = 0
   AND NextLocalAttemptAt IS NULL AND CompletedAt IS NULL`,
		StateUncertain, "dispatch_lease_expired", now, now, job.RequestID,
		job.OriginatingUserID, StateDispatching, job.Fence, job.ClaimToken,
		job.LeaseExpiresAt, now, job.DispatchStartedAt)
	if err != nil {
		return err
	}
	won, err := exactlyOne(result)
	if err != nil {
		return err
	}
	if !won {
		return ErrOwnershipLost
	}
	job.State = StateUncertain
	job.FailureCode = "dispatch_lease_expired"
	job.ClaimToken = ""
	job.LeaseExpiresAt = nil
	job.CompletedAt = &now
	job.LastModifiedDate = now
	return nil
}

func (r *sqlRepository) withTx(ctx context.Context, fn func(*sql.Tx) error) (err error) {
	tx, cleanup, err := sqltx.BeginPessimisticTx(ctx, r.db)
	if err != nil {
		return err
	}
	defer func() {
		_ = tx.Rollback()
		if cleanupErr := cleanup(); cleanupErr != nil {
			err = errors.Join(err, errTransactionCleanup)
		}
	}()
	if err := fn(tx); err != nil {
		return err
	}
	return tx.Commit()
}

func lockUser(ctx context.Context, tx *sql.Tx, userID string) error {
	var id string
	err := tx.QueryRowContext(ctx, `SELECT Id FROM User WHERE Id = ? LIMIT 1 FOR UPDATE`, userID).Scan(&id)
	if errors.Is(err, sql.ErrNoRows) {
		return ErrRequestConflict
	}
	return err
}

const jobSelect = `
SELECT JobId, ConnectedPrincipalId, PrincipalGeneration, OriginatingUserId, State,
       PayloadSha256, PayloadEncodedSize, SourceObjectApiName, SourceRecordId,
       CompletionSnapshot, ClaimToken, LeaseExpiresAt, Fence, DispatchStartedAt,
       AcceptedAt, RemoteReceiptId, RemoteConversationId, ResponseStatus,
       ResultEvidence, FailureCode, LocalAttemptCount, NextLocalAttemptAt,
       CompletedAt, CreatedDate, LastModifiedDate
  FROM _System_ConnectedDeliveryJob`

func loadJob(ctx context.Context, tx *sql.Tx, requestID string, forUpdate bool) (*Job, error) {
	suffix := ` WHERE JobId = ? LIMIT 1`
	if forUpdate {
		suffix += ` FOR UPDATE`
	}
	return scanJob(tx.QueryRowContext(ctx, jobSelect+suffix, requestID))
}

func loadJobForUser(ctx context.Context, tx *sql.Tx, requestID, userID string, forUpdate bool) (*Job, error) {
	suffix := ` WHERE JobId = ? AND OriginatingUserId = ? LIMIT 1`
	if forUpdate {
		suffix += ` FOR UPDATE`
	}
	return scanJob(tx.QueryRowContext(ctx, jobSelect+suffix, requestID, userID))
}

type rowScanner interface {
	Scan(...any) error
}

func scanJob(row rowScanner) (*Job, error) {
	var job Job
	var completion, resultEvidence []byte
	var claim, receipt, conversation, failure sql.NullString
	var lease, dispatched, accepted, nextLocalAttempt, completed sql.NullTime
	var response sql.NullInt64
	if err := row.Scan(&job.RequestID, &job.ConnectedPrincipalID, &job.PrincipalGeneration,
		&job.OriginatingUserID, &job.State, &job.PayloadSHA256, &job.PayloadEncodedSize,
		&job.SourceObject, &job.SourceRecordID, &completion, &claim, &lease, &job.Fence,
		&dispatched, &accepted, &receipt, &conversation, &response, &resultEvidence,
		&failure, &job.LocalAttemptCount, &nextLocalAttempt, &completed,
		&job.CreatedDate, &job.LastModifiedDate); err != nil {
		return nil, err
	}
	job.CompletionSnapshot = append(json.RawMessage(nil), completion...)
	job.ResultEvidence = append(json.RawMessage(nil), resultEvidence...)
	job.ClaimToken = claim.String
	job.RemoteReceiptID = receipt.String
	job.RemoteConversationID = conversation.String
	job.FailureCode = failure.String
	if lease.Valid {
		job.LeaseExpiresAt = &lease.Time
	}
	if dispatched.Valid {
		job.DispatchStartedAt = &dispatched.Time
	}
	if accepted.Valid {
		job.AcceptedAt = &accepted.Time
	}
	if response.Valid {
		value := int(response.Int64)
		job.ResponseStatus = &value
	}
	if nextLocalAttempt.Valid {
		job.NextLocalAttemptAt = &nextLocalAttempt.Time
	}
	if completed.Valid {
		job.CompletedAt = &completed.Time
	}
	if err := validateJobShape(job); err != nil {
		return nil, err
	}
	return &job, nil
}

func validateTerminalDispatch(terminal terminalDispatch) error {
	if terminal.ResponseStatus < 0 || terminal.ResponseStatus > 599 ||
		len(terminal.RemoteReceiptID) > 255 || len(terminal.RemoteConversationID) > 255 ||
		len(terminal.FailureCode) > 64 {
		return ErrInvalidInput
	}
	switch terminal.State {
	case StateAccepted:
		if terminal.RemoteReceiptID == "" || terminal.ResponseStatus < 200 ||
			terminal.ResponseStatus >= 300 || terminal.FailureCode != "" {
			return ErrInvalidInput
		}
	case StateFailed:
		if terminal.RemoteReceiptID != "" || terminal.RemoteConversationID != "" || terminal.FailureCode == "" {
			return ErrInvalidInput
		}
	case StateUncertain:
		if terminal.FailureCode == "" {
			return ErrInvalidInput
		}
	default:
		return ErrInvalidInput
	}
	return nil
}

func exactlyOne(result sql.Result) (bool, error) {
	rows, err := result.RowsAffected()
	return rows == 1, err
}

func nullableString(value string) any {
	if strings.TrimSpace(value) == "" {
		return nil
	}
	return strings.TrimSpace(value)
}

func nullableInt(value int) any {
	if value == 0 {
		return nil
	}
	return value
}

func classifyConflict(err error) error {
	if err == nil {
		return nil
	}
	message := strings.ToLower(err.Error())
	if strings.Contains(message, "duplicate") || strings.Contains(message, "1062") {
		return ErrRequestConflict
	}
	return err
}

func schemaTime(value time.Time) time.Time {
	return value.UTC().Truncate(time.Microsecond)
}