oikos/internal/httpapi/entity_mutations.go

package httpapi

import (
	"context"
	"crypto/sha256"
	"encoding/json"
	"errors"
	"fmt"
	"strconv"
	"strings"

	"github.com/dtoro/oikos/internal/db"
	"github.com/dtoro/oikos/internal/db/sqlcgen"
	"github.com/dtoro/oikos/internal/domain"
	"github.com/dtoro/oikos/internal/httpapi/gen"
	"github.com/dtoro/oikos/internal/observability"
	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
)

func (s *Server) CreateEntity(ctx context.Context, req gen.CreateEntityRequestObject) (gen.CreateEntityResponseObject, error) {
	if req.Body == nil {
		return nil, fmt.Errorf("%w: request body is required", domain.ErrInvalidInput)
	}

	// Check idempotency if a key was provided. The idempotency scope is the
	// calling actor, so replays are per-caller.
	actorType, actorLabel := actorInfo(ctx)
	actor := actorLabel
	var bodyHash string
	if req.Params.IdempotencyKey != nil && *req.Params.IdempotencyKey != "" {
		key := *req.Params.IdempotencyKey
		q := sqlcgen.New(s.pool)
		cached, err := q.GetIdempotentResponse(ctx, sqlcgen.GetIdempotentResponseParams{
			Actor: actor,
			Key:   key,
		})
		if err == nil {
			// Verify the request body hasn't changed.
			bodyJSON, _ := json.Marshal(req.Body)
			bodyHash = fmt.Sprintf("%x", sha256.Sum256(bodyJSON))
			if cached.RequestHash != bodyHash {
				return nil, fmt.Errorf("%w: idempotency key %s used with different request body", domain.ErrConflict, key)
			}
			// Replay the cached response.
			if cached.ResponseCode != nil && *cached.ResponseCode == 201 {
				var entity gen.Entity
				if len(cached.ResponseBody) > 0 {
					if err := json.Unmarshal(cached.ResponseBody, &entity); err != nil {
						return nil, fmt.Errorf("unmarshal cached response: %w", err)
					}
				}
				return gen.CreateEntity201JSONResponse{
					Body:    entity,
					Headers: gen.CreateEntity201ResponseHeaders{ETag: `"` + strconv.Itoa(entity.Version) + `"`},
				}, nil
			}
			// Forward cached error response.
			return gen.CreateEntitydefaultApplicationProblemPlusJSONResponse{
				Body:       gen.Problem{Status: int(*cached.ResponseCode), Title: "replayed error"},
				StatusCode: int(*cached.ResponseCode),
			}, nil
		}
	}

	id, err := uuid.NewV7()
	if err != nil {
		return nil, err
	}
	slug := req.Body.Slug
	if slug == "" {
		slug = req.Body.Type + ":" + req.Body.Name
	}

	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback(ctx)
	q := sqlcgen.New(tx)

	// Validate type exists and is NOT abstract.
	var isAbstract bool
	if err := tx.QueryRow(ctx, `SELECT is_abstract FROM entity_types WHERE name = $1`, req.Body.Type).Scan(&isAbstract); err != nil {
		if err == pgx.ErrNoRows {
			return nil, fmt.Errorf("%w: entity type %q", domain.ErrNotFound, req.Body.Type)
		}
		return nil, err
	}
	if isAbstract {
		return nil, fmt.Errorf("%w: %s", domain.ErrAbstractType, req.Body.Type)
	}

	// Get default state from lifecycle.
	var defaultState *string
	var lcDefault string
	if err := tx.QueryRow(ctx, `SELECT ld.default_state FROM lifecycle_defs ld
		JOIN entity_types et ON et.lifecycle_id = ld.id
		WHERE et.name = $1`, req.Body.Type).Scan(&lcDefault); err == nil {
		defaultState = &lcDefault
	}

	state := req.Body.State
	if state == nil && defaultState != nil {
		state = defaultState
	}

	// attributes is NOT NULL; the column default only applies when omitted,
	// not when an explicit NULL is bound — so default to an empty object.
	attrsJSON := []byte("{}")
	if req.Body.Attributes != nil {
		attrsJSON, _ = json.Marshal(req.Body.Attributes)
	}

	// Insert the entity.
	inserted, err := q.InsertEntity(ctx, sqlcgen.InsertEntityParams{
		ID:         id,
		Slug:       slug,
		Type:       req.Body.Type,
		Name:       req.Body.Name,
		State:      state,
		Attributes: attrsJSON,
	})
	if err != nil {
		// Duplicate slug.
		if strings.Contains(err.Error(), "unique") || strings.Contains(err.Error(), "duplicate") {
			return nil, fmt.Errorf("%w: slug %q already exists", domain.ErrAlreadyExists, slug)
		}
		return nil, err
	}

	// Convert sqlcgen.Entity → gen.Entity.
	entity := sqlcEntityToGen(inserted)

	// Cache idempotent response.
	if req.Params.IdempotencyKey != nil && *req.Params.IdempotencyKey != "" {
		respBody, _ := json.Marshal(entity)
		code := int32(201)
		if bodyHash == "" {
			bodyJSON, _ := json.Marshal(req.Body)
			bodyHash = fmt.Sprintf("%x", sha256.Sum256(bodyJSON))
		}
		if putErr := q.PutIdempotentResponse(ctx, sqlcgen.PutIdempotentResponseParams{
			Actor:        actor,
			Key:          *req.Params.IdempotencyKey,
			RequestHash:  bodyHash,
			ResponseCode: &code,
			ResponseBody: respBody,
		}); putErr != nil {
			return nil, putErr
		}
	}

	// Audit.
	entityID := inserted.ID
	if auditErr := observability.Audit(ctx, q, actorType, actor, "create",
		&entityID, "POST", "/api/v1/entities", "",
		nil,
		map[string]any{"type": req.Body.Type, "slug": slug}); auditErr != nil {
		return nil, auditErr
	}

	if eventErr := observability.Event(ctx, q, "entity.created", &entityID,
		"info", "oikos-api", "",
		map[string]any{"slug": slug, "type": req.Body.Type}); eventErr != nil {
		return nil, eventErr
	}

	if err := ensureDefaultChecks(ctx, tx, inserted.ID, slug, req.Body.Type, inserted.Name, attrsJSON); err != nil {
		return nil, err
	}

	if err := tx.Commit(ctx); err != nil {
		return nil, err
	}

	return gen.CreateEntity201JSONResponse{
		Body:    entity,
		Headers: gen.CreateEntity201ResponseHeaders{ETag: `"` + strconv.Itoa(entity.Version) + `"`},
	}, nil
}

func (s *Server) PatchEntity(ctx context.Context, req gen.PatchEntityRequestObject) (gen.PatchEntityResponseObject, error) {
	if req.Body == nil {
		return nil, fmt.Errorf("%w: request body is required", domain.ErrInvalidInput)
	}

	id, err := s.resolveEntityID(ctx, req.Id)
	if err != nil {
		return nil, err
	}

	// Parse If-Match header (quoted version string).
	ifMatch := strings.Trim(req.Params.IfMatch, `"`)
	expectedVersion, err := strconv.Atoi(ifMatch)
	if err != nil {
		return nil, fmt.Errorf("%w: invalid If-Match header %q", domain.ErrInvalidInput, req.Params.IfMatch)
	}

	tx, err := s.pool.Begin(ctx)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback(ctx)

	// Get current entity for version check + lifecycle validation.
	current, err := sqlcgen.New(tx).GetEntityByID(ctx, id)
	if err != nil {
		if err == pgx.ErrNoRows {
			return nil, fmt.Errorf("%w: %s", domain.ErrNotFound, req.Id)
		}
		return nil, err
	}

	if int(current.Version) != expectedVersion {
		return nil, fmt.Errorf("%w: expected version %d, current version %d",
			domain.ErrConflict, expectedVersion, current.Version)
	}

	// Validate lifecycle transition if state is being changed.
	if req.Body.State != nil && *req.Body.State != "" {
		fromState := ""
		if current.State != nil {
			fromState = *current.State
		}
		if err := db.ValidateTransition(ctx, tx, id, current.Type, fromState, *req.Body.State); err != nil {
			if errors.Is(err, db.ErrTransitionInvalid) {
				return nil, fmt.Errorf("%w: %v", domain.ErrInvalidTransition, err)
			}
			return nil, err
		}
	}

	// Check idempotency (note: the spec doesn't define Idempotency-Key for PATCH,
	// but we handle it if the generated code ever adds it).
	// For now, no idempotency check on PATCH.

	// Marshal attributes if provided.
	var attrsJSON []byte
	if req.Body.Attributes != nil {
		attrsJSON, _ = json.Marshal(req.Body.Attributes)
	}

	// Perform the update via sqlcgen.
	q := sqlcgen.New(tx)
	updated, err := q.UpdateEntity(ctx, sqlcgen.UpdateEntityParams{
		Name:             req.Body.Name,
		State:            req.Body.State,
		Attributes:       attrsJSON,
		SetMaintenance:   req.Body.MaintenanceUntil != nil,
		MaintenanceUntil: req.Body.MaintenanceUntil,
		ID:               id,
		Version:          int32(expectedVersion),
	})
	if err != nil {
		if err == pgx.ErrNoRows {
			// Version mismatch or entity not found.
			return nil, fmt.Errorf("%w: entity was modified concurrently", domain.ErrConflict)
		}
		return nil, err
	}

	entity := sqlcEntityToGen(updated)

	// Audit.
	patchActorType, patchActor := actorInfo(ctx)
	if auditErr := observability.Audit(ctx, q, patchActorType, patchActor, "patch",
		&id, "PATCH", "/api/v1/entities/"+req.Id, "",
		nil,
		map[string]any{"version": expectedVersion}); auditErr != nil {
		return nil, auditErr
	}

	if eventErr := observability.Event(ctx, q, "entity.updated", &id,
		"info", "oikos-api", "",
		map[string]any{"slug": entity.Slug, "type": entity.Type, "version": updated.Version}); eventErr != nil {
		return nil, eventErr
	}

	if err := tx.Commit(ctx); err != nil {
		return nil, err
	}

	s.entityCache.Invalidate(entity.Slug, entity.Id.String())

	return gen.PatchEntity200JSONResponse{
		Body:    entity,
		Headers: gen.PatchEntity200ResponseHeaders{ETag: `"` + strconv.Itoa(entity.Version) + `"`},
	}, nil
}