query-orchestration/internal/fieldextraction/service.go

package fieldextraction

import (
	"context"
	"errors"
	"fmt"

	"queryorchestration/internal/database/repository"
	"queryorchestration/internal/serviceconfig"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
)

// ErrDocumentNotFound is returned by CreateFieldExtraction when the
// target document does not exist. Handlers map this sentinel to 404.
var ErrDocumentNotFound = errors.New("document not found")

// ErrMutualExclusivityViolation is returned by CreateFieldExtraction
// when the target document is already bound to a custom schema via
// documents.custom_schema_id. The mutable-metadata feature disallows
// mixing legacy field extractions with the new metadata system on the
// same document. Handlers map this sentinel to 409.
var ErrMutualExclusivityViolation = errors.New("document is bound to a custom schema and cannot accept legacy field extractions")

// Service provides field extraction operations
type Service struct {
	cfg serviceconfig.ConfigProvider
}

// New creates a new field extraction service
func New(cfg serviceconfig.ConfigProvider) *Service {
	return &Service{cfg: cfg}
}

// CreateFieldExtraction creates a new field extraction with all single and array fields
// in a single transaction. All array fields must have consistent length.
// Returns the created field extraction record.
func (s *Service) CreateFieldExtraction(ctx context.Context, input *CreateFieldExtractionInput) (*repository.Documentfieldextraction, error) {
	if err := validateCreateFieldExtractionInput(input); err != nil {
		return nil, err
	}

	// Begin transaction
	tx, err := s.cfg.GetDBPool().Begin(ctx)
	if err != nil {
		return nil, fmt.Errorf("failed to begin transaction: %w", err)
	}
	defer func() {
		_ = tx.Rollback(ctx)
	}()

	queries := s.cfg.GetDBQueries().WithTx(tx)

	// Milestone 2.5: take the parent-row lock on documents as the FIRST
	// DML inside the tx. Serializes this write with any concurrent
	// AssignSchema (which holds the same FOR UPDATE lock) so the
	// mutual-exclusivity invariant cannot be broken under READ COMMITTED
	// (see plan Section 7.3). A missing row → ErrDocumentNotFound; a
	// non-null custom_schema_id on the locked row →
	// ErrMutualExclusivityViolation. Both sentinels surface BEFORE any
	// other DML so the 404/409 paths leave the DB unchanged.
	lockedRow, err := queries.LockDocumentForLegacyExtractionWrite(ctx, input.DocumentID)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, ErrDocumentNotFound
		}
		return nil, fmt.Errorf("failed to lock document for legacy extraction: %w", err)
	}
	if lockedRow.CustomSchemaID != nil {
		return nil, ErrMutualExclusivityViolation
	}

	// Set document ID in single fields
	input.SingleFields.Documentid = input.DocumentID

	// Create main field extraction record with single fields
	fieldExtraction, err := queries.AddFieldExtraction(ctx, input.SingleFields)
	if err != nil {
		return nil, fmt.Errorf("failed to create field extraction: %w", err)
	}

	// Lock existing version rows for this document to prevent concurrent version creation
	// This ensures that two concurrent requests cannot create the same version number
	_, err = queries.LockFieldExtractionVersionsForDocument(ctx, input.DocumentID)
	if err != nil {
		return nil, fmt.Errorf("failed to lock version rows: %w", err)
	}

	// Get the max version number for this document (after locking)
	// COALESCE in the query ensures we get 0 if no versions exist
	maxVersionPtr, err := queries.GetMaxFieldExtractionVersion(ctx, input.DocumentID)
	if err != nil {
		return nil, fmt.Errorf("failed to get max version: %w", err)
	}
	// SQLC returns *int64 even though COALESCE guarantees non-NULL
	// Dereference safely - if somehow nil, treat as 0
	var maxVersion int64
	if maxVersionPtr != nil {
		maxVersion = *maxVersionPtr
	}
	// Next version is always maxVersion + 1 (1 for first version since maxVersion is 0)
	version := maxVersion + 1

	// Create version entry with documentId for the unique constraint
	err = queries.AddFieldExtractionEntry(ctx, &repository.AddFieldExtractionEntryParams{
		Fieldextractionid: fieldExtraction.ID,
		Documentid:        input.DocumentID,
		Version:           version,
		Createdby:         input.SingleFields.Createdby,
	})
	if err != nil {
		return nil, fmt.Errorf("failed to create field extraction version entry: %w", err)
	}

	// Insert all array field rows
	for arrayIndex, arrayField := range input.ArrayFields {
		// Set field extraction ID and array index
		arrayField.Fieldextractionid = fieldExtraction.ID
		//nolint:gosec // Array index is expected to be small, int16 is sufficient
		arrayField.Arrayindex = int16(arrayIndex)

		err = queries.AddFieldExtractionArrayField(ctx, arrayField)
		if err != nil {
			return nil, fmt.Errorf("failed to create array field at index %d: %w", arrayIndex, err)
		}
	}

	// Commit transaction
	if err := tx.Commit(ctx); err != nil {
		return nil, fmt.Errorf("failed to commit transaction: %w", err)
	}

	return fieldExtraction, nil
}

// GetCurrentFieldExtraction retrieves the most recent field extraction for a document
// Returns nil if no field extraction exists for the document
func (s *Service) GetCurrentFieldExtraction(ctx context.Context, documentID uuid.UUID) (*repository.Currentfieldextraction, error) {
	if documentID == uuid.Nil {
		return nil, fmt.Errorf("documentID cannot be nil")
	}

	fieldExtraction, err := s.cfg.GetDBQueries().GetCurrentFieldExtraction(ctx, documentID)
	if err != nil {
		if err == pgx.ErrNoRows {
			return nil, nil
		}
		return nil, fmt.Errorf("failed to get current field extraction: %w", err)
	}

	return fieldExtraction, nil
}

// GetFieldExtractionHistory retrieves all field extraction versions for a document
// ordered by version (most recent first)
func (s *Service) GetFieldExtractionHistory(ctx context.Context, documentID uuid.UUID) ([]*repository.GetFieldExtractionHistoryRow, error) {
	if documentID == uuid.Nil {
		return nil, fmt.Errorf("documentID cannot be nil")
	}

	history, err := s.cfg.GetDBQueries().GetFieldExtractionHistory(ctx, documentID)
	if err != nil {
		return nil, fmt.Errorf("failed to get field extraction history: %w", err)
	}

	return history, nil
}

// GetFieldExtractionArrayFields retrieves all array fields for a specific field extraction
// ordered by arrayIndex
func (s *Service) GetFieldExtractionArrayFields(ctx context.Context, fieldExtractionID uuid.UUID) ([]*repository.Documentfieldextractionarrayfield, error) {
	if fieldExtractionID == uuid.Nil {
		return nil, fmt.Errorf("fieldExtractionID cannot be nil")
	}

	arrayFields, err := s.cfg.GetDBQueries().GetFieldExtractionArrayFields(ctx, fieldExtractionID)
	if err != nil {
		return nil, fmt.Errorf("failed to get array fields: %w", err)
	}

	return arrayFields, nil
}

// GetFieldExtractionByVersion retrieves a specific version of field extraction for a document
// Returns nil if no field extraction exists for the document at that version
// documentID: the UUID of the document
// version: the version number to retrieve (must be >= 1)
func (s *Service) GetFieldExtractionByVersion(ctx context.Context, documentID uuid.UUID, version int64) (*repository.GetFieldExtractionByVersionRow, error) {
	if documentID == uuid.Nil {
		return nil, fmt.Errorf("documentID cannot be nil")
	}

	if version < 1 {
		return nil, fmt.Errorf("version must be at least 1")
	}

	fieldExtraction, err := s.cfg.GetDBQueries().GetFieldExtractionByVersion(ctx, &repository.GetFieldExtractionByVersionParams{
		Documentid: documentID,
		Version:    version,
	})
	if err != nil {
		if err == pgx.ErrNoRows {
			return nil, nil
		}
		return nil, fmt.Errorf("failed to get field extraction by version: %w", err)
	}

	return fieldExtraction, nil
}

// validateCreateFieldExtractionInput checks the basic shape invariants of
// the CreateFieldExtraction input before any DB work begins. Extracted so
// CreateFieldExtraction stays under the cyclomatic complexity ceiling.
func validateCreateFieldExtractionInput(input *CreateFieldExtractionInput) error {
	if input == nil {
		return fmt.Errorf("input cannot be nil")
	}
	if input.DocumentID == uuid.Nil {
		return fmt.Errorf("documentID cannot be nil")
	}
	if input.SingleFields == nil {
		return fmt.Errorf("singleFields cannot be nil")
	}
	if input.SingleFields.Createdby == "" {
		return fmt.Errorf("createdBy cannot be empty")
	}
	if _, err := ValidateArrayConsistency(input.ArrayFields); err != nil {
		return fmt.Errorf("array validation failed: %w", err)
	}
	return nil
}