query-orchestration/internal/customschema/service.go

package customschema

import (
	"context"
	"errors"
	"fmt"
	"time"

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

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

// ErrSchemaNotFound is the typed sentinel returned when a schema lookup
// cannot find a row. The service layer maps pgx.ErrNoRows (and any other
// "row is missing" condition) onto this so HTTP handlers can translate a
// single sentinel into 404 without peeking at driver errors.
var ErrSchemaNotFound = errors.New("schema not found")

// ErrSchemaInUse is the typed sentinel returned when a write operation is
// rejected because the schema is still referenced by one or more documents
// via documents.custom_schema_id. HTTP handlers map this to 409.
var ErrSchemaInUse = errors.New("schema is in use")

// ErrSchemaNameConflict is the typed sentinel returned when CreateSchema is
// asked to insert a (client_id, name, version) tuple that already exists.
// HTTP handlers map this to 409. Surfaces whenever the unique constraint
// uq_client_schema_name_version rejects the insert.
var ErrSchemaNameConflict = errors.New("schema name already exists for client")

// ErrSchemaParentNotActive is the typed sentinel returned when
// CreateSchemaVersion is asked to branch from a parent row whose status
// is not 'active' (but is also not 'retired'). The canonical case is
// "caller passed v1's id while v2 is the current active row for the
// lineage". HTTP handlers map this to 409.
var ErrSchemaParentNotActive = errors.New("parent schema is not the active version in its lineage")

// ErrSchemaParentRetired is the typed sentinel returned when
// CreateSchemaVersion is asked to branch from a parent row whose status
// is 'retired'. It is distinct from ErrSchemaParentNotActive so the
// handler layer can surface a more specific reason. HTTP handlers map
// this to 409.
var ErrSchemaParentRetired = errors.New("parent schema is retired")

// defaultListLimit is applied when a caller leaves ListFilters.Limit at
// its zero value so the underlying SQL LIMIT clause never receives 0.
const defaultListLimit int64 = 100

// maxListLimit is the hard ceiling on list results, matching the OpenAPI
// spec's maximum: 200 constraint. A service-layer backstop ensures no
// caller (including future non-HTTP paths) can exceed this value.
const maxListLimit int64 = 200

// Service owns schema CRUD and (later) metadata operations for the
// mutable-metadata feature. It holds the database-backed config provider,
// the schema validator, and an audit sink. See plan Section 7.1.
type Service struct {
	cfg       serviceconfig.ConfigProvider
	validator *SchemaValidator
	audit     AuditSink
}

// New constructs a Service. The validator and audit sink are injected so
// tests can substitute an in-memory capturing sink without mocking.
func New(cfg serviceconfig.ConfigProvider, validator *SchemaValidator, audit AuditSink) *Service {
	return &Service{cfg: cfg, validator: validator, audit: audit}
}

// CreateSchema validates the supplied schema_def, inserts a version 1 row
// with status='active', and emits a schema.create audit entry on success.
// Validation failures and unique-constraint rejections do NOT emit audit
// entries. The actor argument is the authenticated Cognito subject — it is
// never read from the input struct.
func (s *Service) CreateSchema(ctx context.Context, input CreateSchemaInput, actor string) (*Schema, error) {
	if err := s.validator.ValidateSchemaDefinition(input.SchemaDef); err != nil {
		return nil, fmt.Errorf("customschema: validate schema definition: %w", err)
	}

	queries := s.cfg.GetDBQueries()
	description := input.Description
	params := &repository.CreateClientMetadataSchemaParams{
		ClientID:    input.ClientID,
		Name:        input.Name,
		Description: &description,
		SchemaDef:   []byte(input.SchemaDef),
		Version:     1,
		Status:      repository.SchemaStatusTypeActive,
		CreatedBy:   actor,
	}

	row, err := queries.CreateClientMetadataSchema(ctx, params)
	if err != nil {
		var pgErr *pgconn.PgError
		if errors.As(err, &pgErr) && pgErr.Code == "23505" &&
			pgErr.ConstraintName == "uq_client_schema_name_version" {
			return nil, ErrSchemaNameConflict
		}
		return nil, fmt.Errorf("customschema: create schema: %w", err)
	}

	rec := AuditRecord{
		Actor:      actor,
		Action:     AuditActionSchemaCreate,
		ResourceID: row.ID,
		ClientID:   row.ClientID,
		Timestamp:  time.Now().UTC(),
		Details: map[string]any{
			"name":    row.Name,
			"version": row.Version,
		},
	}
	if auditErr := s.audit.Record(ctx, rec); auditErr != nil {
		return nil, fmt.Errorf("customschema: record create audit: %w", auditErr)
	}

	return rowToSchema(row), nil
}

// CreateSchemaVersion validates the supplied schema_def, then inside a
// single transaction locks the (client_id, name) lineage with FOR UPDATE,
// re-reads the parent row under the lock, verifies the parent is still
// 'active', inserts a new row at max(version)+1 with status='active', and
// flips the parent row to 'superseded'. On success it emits a schema.update
// audit entry whose ResourceID is the new version row's id. Missing parent
// surfaces as ErrSchemaNotFound. Non-active-but-not-retired parents surface
// as ErrSchemaParentNotActive. Retired parents surface as
// ErrSchemaParentRetired. The actor argument is the authenticated Cognito
// subject — it is never read from the input struct.
func (s *Service) CreateSchemaVersion(ctx context.Context, parentSchemaID uuid.UUID, input CreateSchemaVersionInput, actor string) (*Schema, error) {
	// 1. Fail fast on an invalid schema_def before opening any tx.
	if err := s.validator.ValidateSchemaDefinition(input.SchemaDef); err != nil {
		return nil, fmt.Errorf("customschema: validate schema definition: %w", err)
	}

	// 2. Outer-read the parent to get (client_id, name) so the tx can lock
	//    the right lineage. Missing row maps to ErrSchemaNotFound.
	parent, err := s.cfg.GetDBQueries().GetClientMetadataSchema(ctx, parentSchemaID)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return nil, ErrSchemaNotFound
		}
		return nil, fmt.Errorf("customschema: load parent schema: %w", err)
	}

	// 3. Run the write inside a transaction. The closure captures newRow
	//    so we can emit the audit entry after a successful commit.
	var newRow *repository.ClientMetadataSchema
	txErr := s.cfg.ExecuteDBTransaction(ctx, func(txCtx context.Context, q *repository.Queries) error {
		// 3a. Lock every row for (client_id, name). This serializes with
		//     any concurrent CreateSchemaVersion for the same lineage:
		//     the second caller blocks here until the first commits, then
		//     reads the now-superseded parent and bails out cleanly.
		locked, lockErr := q.LockSchemaVersionsForName(txCtx, &repository.LockSchemaVersionsForNameParams{
			ClientID: parent.ClientID,
			Name:     parent.Name,
		})
		if lockErr != nil {
			return fmt.Errorf("customschema: lock schema versions: %w", lockErr)
		}

		// 3b. Re-read the parent row from the locked set to get its
		//     authoritative status. Also track max(version) in the same
		//     pass so we don't need a second query.
		var parentRow *repository.ClientMetadataSchema
		var maxVersion int32
		for _, r := range locked {
			if r.ID == parentSchemaID {
				parentRow = r
			}
			if r.Version > maxVersion {
				maxVersion = r.Version
			}
		}
		if parentRow == nil {
			return ErrSchemaNotFound
		}

		// 3c. Parent status gate. Retired is checked first so it gets the
		//     more specific sentinel; any other non-active status (in
		//     practice 'superseded') maps to ErrSchemaParentNotActive.
		switch parentRow.Status {
		case repository.SchemaStatusTypeRetired:
			return ErrSchemaParentRetired
		case repository.SchemaStatusTypeActive:
			// proceed
		default:
			return ErrSchemaParentNotActive
		}

		// 3d. Insert the new version row at max+1.
		description := input.Description
		insertParams := &repository.CreateClientMetadataSchemaParams{
			ClientID:    parent.ClientID,
			Name:        parent.Name,
			Description: &description,
			SchemaDef:   []byte(input.SchemaDef),
			Version:     maxVersion + 1,
			Status:      repository.SchemaStatusTypeActive,
			CreatedBy:   actor,
		}
		inserted, insertErr := q.CreateClientMetadataSchema(txCtx, insertParams)
		if insertErr != nil {
			return fmt.Errorf("customschema: insert new version: %w", insertErr)
		}

		// 3e. Supersede the parent row inside the same tx so the
		//     active-parent invariant holds at commit time.
		if statusErr := q.SetSchemaStatus(txCtx, &repository.SetSchemaStatusParams{
			ID:     parentSchemaID,
			Status: repository.SchemaStatusTypeSuperseded,
		}); statusErr != nil {
			return fmt.Errorf("customschema: supersede parent: %w", statusErr)
		}

		newRow = inserted
		return nil
	})
	if txErr != nil {
		return nil, txErr
	}

	// 4. Audit on success only. ResourceID is the new version's id per
	//    the writes_schema_update_audit_entry subtest contract.
	rec := AuditRecord{
		Actor:      actor,
		Action:     AuditActionSchemaUpdate,
		ResourceID: newRow.ID,
		ClientID:   newRow.ClientID,
		Timestamp:  time.Now().UTC(),
		Details: map[string]any{
			"name":             newRow.Name,
			"version":          newRow.Version,
			"parent_schema_id": parentSchemaID.String(),
		},
	}
	if auditErr := s.audit.Record(ctx, rec); auditErr != nil {
		return nil, fmt.Errorf("customschema: record update audit: %w", auditErr)
	}

	return rowToSchema(newRow), nil
}

// GetSchema loads a single schema by primary key, decorated with its live
// document count so callers receive accurate DocumentCount and CanDelete
// values. Missing rows surface as ErrSchemaNotFound.
func (s *Service) GetSchema(ctx context.Context, schemaID uuid.UUID) (SchemaSummary, error) {
	queries := s.cfg.GetDBQueries()
	row, err := queries.GetClientMetadataSchema(ctx, schemaID)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return SchemaSummary{}, ErrSchemaNotFound
		}
		return SchemaSummary{}, fmt.Errorf("customschema: get schema: %w", err)
	}
	count, err := queries.GetSchemaDocumentCount(ctx, &schemaID)
	if err != nil {
		return SchemaSummary{}, fmt.Errorf("customschema: get schema document count: %w", err)
	}
	return SchemaSummary{
		Schema:        *rowToSchema(row),
		DocumentCount: count,
		CanDelete:     count == 0,
	}, nil
}

// ListSchemas translates ListFilters onto the generated sqlc parameter
// struct, runs the query, and decorates each row with DocumentCount and
// CanDelete. CanDelete is true iff DocumentCount == 0, mirroring plan
// Section 7.1.
func (s *Service) ListSchemas(ctx context.Context, clientID string, filters ListFilters) ([]SchemaSummary, error) {
	queries := s.cfg.GetDBQueries()

	var nameFilter *string
	if filters.Name != "" {
		name := filters.Name
		nameFilter = &name
	}

	statusFilter := repository.SchemaStatusTypeActive
	if !filters.StatusAny && filters.Status != "" {
		statusFilter = repository.SchemaStatusType(filters.Status)
	}

	limit := int64(filters.Limit)
	switch {
	case limit <= 0:
		limit = defaultListLimit
	case limit > maxListLimit:
		limit = maxListLimit
	}

	params := &repository.ListClientMetadataSchemasParams{
		ClientID:           clientID,
		NameFilter:         nameFilter,
		IncludeAllStatuses: filters.StatusAny,
		StatusFilter:       statusFilter,
		IncludeAllVersions: filters.IncludeAllVersions,
		LimitVal:           limit,
		OffsetVal:          int64(filters.Offset),
	}

	rows, err := queries.ListClientMetadataSchemas(ctx, params)
	if err != nil {
		return nil, fmt.Errorf("customschema: list schemas: %w", err)
	}

	summaries := make([]SchemaSummary, 0, len(rows))
	for _, row := range rows {
		summaries = append(summaries, listRowToSummary(row))
	}
	return summaries, nil
}

// DeleteSchema retires (does not physically delete) a schema. It refuses
// to retire a schema that still has documents bound to it via
// documents.custom_schema_id and surfaces ErrSchemaInUse in that case.
// Missing rows surface as ErrSchemaNotFound. On success it emits a
// schema.delete audit entry.
func (s *Service) DeleteSchema(ctx context.Context, schemaID uuid.UUID, actor string) error {
	queries := s.cfg.GetDBQueries()

	row, err := queries.GetClientMetadataSchema(ctx, schemaID)
	if err != nil {
		if errors.Is(err, pgx.ErrNoRows) {
			return ErrSchemaNotFound
		}
		return fmt.Errorf("customschema: load schema for delete: %w", err)
	}

	count, err := queries.GetSchemaDocumentCount(ctx, &schemaID)
	if err != nil {
		return fmt.Errorf("customschema: count documents for delete: %w", err)
	}
	if count > 0 {
		return ErrSchemaInUse
	}

	if err := queries.SetSchemaStatus(ctx, &repository.SetSchemaStatusParams{
		ID:     schemaID,
		Status: repository.SchemaStatusTypeRetired,
	}); err != nil {
		return fmt.Errorf("customschema: retire schema: %w", err)
	}

	rec := AuditRecord{
		Actor:      actor,
		Action:     AuditActionSchemaDelete,
		ResourceID: schemaID,
		ClientID:   row.ClientID,
		Timestamp:  time.Now().UTC(),
		Details: map[string]any{
			"name":    row.Name,
			"version": row.Version,
		},
	}
	if auditErr := s.audit.Record(ctx, rec); auditErr != nil {
		return fmt.Errorf("customschema: record delete audit: %w", auditErr)
	}
	return nil
}

// rowToSchema translates a sqlc-generated ClientMetadataSchema row into
// the customschema.Schema model returned to handlers. The pointer-valued
// Description column collapses to "" when nil and the Timestamptz created_at
// is unwrapped via its Time field.
func rowToSchema(row *repository.ClientMetadataSchema) *Schema {
	description := ""
	if row.Description != nil {
		description = *row.Description
	}
	return &Schema{
		ID:          row.ID,
		ClientID:    row.ClientID,
		Name:        row.Name,
		Description: description,
		SchemaDef:   append([]byte(nil), row.SchemaDef...),
		Version:     int(row.Version),
		Status:      string(row.Status),
		CreatedAt:   row.CreatedAt.Time,
		CreatedBy:   row.CreatedBy,
	}
}

// listRowToSummary translates a generated ListClientMetadataSchemasRow
// into a SchemaSummary with CanDelete derived from DocumentCount.
func listRowToSummary(row *repository.ListClientMetadataSchemasRow) SchemaSummary {
	description := ""
	if row.Description != nil {
		description = *row.Description
	}
	return SchemaSummary{
		Schema: Schema{
			ID:          row.ID,
			ClientID:    row.ClientID,
			Name:        row.Name,
			Description: description,
			SchemaDef:   append([]byte(nil), row.SchemaDef...),
			Version:     int(row.Version),
			Status:      string(row.Status),
			CreatedAt:   row.CreatedAt.Time,
			CreatedBy:   row.CreatedBy,
		},
		DocumentCount: row.DocumentCount,
		CanDelete:     row.DocumentCount == 0,
	}
}