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, } }