Files
query-orchestration/internal/customschema/service_metadata.go
T
Jay Brown 17fc813823 Merged in feature/mutable-metadata1 (pull request #221)
M1, M2 and M3 complete

* M1, M2 and M3 complete

* review changes

* docs

* docs
2026-04-16 23:11:26 +00:00

538 lines
20 KiB
Go

package customschema
import (
"context"
"encoding/json"
"errors"
"fmt"
"time"
"queryorchestration/internal/database/repository"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgconn"
)
// ErrDocumentNotFound is the typed sentinel returned when a metadata or
// schema-assign operation targets a document that does not exist. Handlers
// map this to 404.
var ErrDocumentNotFound = errors.New("document not found")
// ErrDocumentNotBoundToSchema is returned when SetDocumentMetadata is
// called on a document whose custom_schema_id is NULL. Handlers map this
// to 400 — the client needs to bind a schema via PATCH
// /super-admin/documents/{id}/schema before writing metadata.
var ErrDocumentNotBoundToSchema = errors.New("document is not bound to a custom schema")
// ErrDocumentHasLegacyExtractions is returned when any mutable-metadata
// write path rejects a document that already carries a
// documentFieldExtractionVersions row. Handlers map this to 409.
var ErrDocumentHasLegacyExtractions = errors.New("document has legacy field extractions")
// ErrDocumentHasCustomMetadata is returned when AssignSchema is asked to
// change (or clear) a schema binding on a document that already has at
// least one document_custom_metadata row. Handlers map this to 409. This
// mirrors the DB-level Trigger 1 guard from migration 130.
var ErrDocumentHasCustomMetadata = errors.New("document already has custom metadata")
// ErrMetadataPayloadTooLarge is returned when SetDocumentMetadata receives
// a payload whose length is strictly greater than MaxMetadataPayloadBytes.
// Handlers map this to 400.
var ErrMetadataPayloadTooLarge = errors.New("metadata payload exceeds maximum size")
// ErrMetadataVersionNotFound is returned when GetMetadataByVersion cannot
// find the requested version. Handlers map this to 404.
var ErrMetadataVersionNotFound = errors.New("metadata version not found")
// ErrSchemaClientMismatch is returned when AssignSchema is asked to bind a
// document to a schema owned by a different client. The composite FK
// fk_documents_custom_schema_same_client enforces this at the DB layer;
// this sentinel carries the violation up to the 409 handler branch.
var ErrSchemaClientMismatch = errors.New("schema and document belong to different clients")
// ErrSchemaNotActive is returned when AssignSchema is asked to bind a
// document to a schema whose status is not 'active'. Handlers map this to
// 409.
var ErrSchemaNotActive = errors.New("schema is not active")
// SetDocumentMetadata writes a new metadata version for a document. The
// write order is:
//
// 1. LockDocumentForMetadataWrite (parent-row lock on documents) — the
// FIRST DML inside the tx so concurrent writers serialize here.
// 2. Read documents.custom_schema_id — reject if NULL (no binding).
// 3. Check for legacy field extractions — reject if present.
// 4. Load schema_def from the bound schema row — used by the validator.
// 5. Validate the payload against the schema (after size ceiling check).
// 6. GetMaxDocumentMetadataVersion then CreateDocumentCustomMetadata at
// version = max + 1.
// 7. Emit metadata.write audit entry post-commit.
//
// The actor argument is the authenticated Cognito subject. It is the only
// source of the created_by column; the input struct has no such field.
func (s *Service) SetDocumentMetadata(ctx context.Context, input SetMetadataInput, actor string) (*CustomMetadata, error) {
// Fail fast on the trivial input rules before opening a tx.
if input.DocumentID == uuid.Nil {
return nil, fmt.Errorf("customschema: document id is required")
}
if len(input.Metadata) > MaxMetadataPayloadBytes {
return nil, ErrMetadataPayloadTooLarge
}
var newRow *repository.DocumentCustomMetadatum
var schemaID *uuid.UUID
var schemaName *string
var schemaVersion *int
var schemaClientID string // captured for the post-commit audit record
runTx := func() error {
return s.cfg.ExecuteDBTransaction(ctx, func(txCtx context.Context, q *repository.Queries) error {
// 1. Parent-row lock. Must be the first DML inside the tx.
if _, err := q.LockDocumentForMetadataWrite(txCtx, input.DocumentID); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return ErrDocumentNotFound
}
return fmt.Errorf("customschema: lock document for metadata write: %w", err)
}
// 2. Read the current binding. NULL means the document has not
// opted in to custom metadata yet.
boundSchemaID, err := q.GetDocumentCustomSchemaId(txCtx, input.DocumentID)
if err != nil {
return fmt.Errorf("customschema: read document schema binding: %w", err)
}
if boundSchemaID == nil {
return ErrDocumentNotBoundToSchema
}
// 3. Legacy extraction guard. The existing HasFieldExtraction
// query checks documentFieldExtractionVersions which is the
// same table the DB trigger uses.
hasLegacy, err := q.HasFieldExtraction(txCtx, input.DocumentID)
if err != nil {
return fmt.Errorf("customschema: check legacy extractions: %w", err)
}
if hasLegacy {
return ErrDocumentHasLegacyExtractions
}
// 4. Load the schema row so we can validate the payload and
// capture schema name/version for the response.
schemaRow, err := q.GetClientMetadataSchema(txCtx, *boundSchemaID)
if err != nil {
return fmt.Errorf("customschema: load bound schema: %w", err)
}
localSchemaID := schemaRow.ID
schemaID = &localSchemaID
localSchemaName := schemaRow.Name
schemaName = &localSchemaName
localSchemaVersion := int(schemaRow.Version)
schemaVersion = &localSchemaVersion
schemaClientID = schemaRow.ClientID
// 5. Validate the metadata against the bound schema.
if err := s.validator.ValidateMetadata(input.Metadata, schemaRow.SchemaDef); err != nil {
return fmt.Errorf("customschema: validate metadata: %w", err)
}
// 6. Compute next version and insert.
maxVersion, err := q.GetMaxDocumentMetadataVersion(txCtx, input.DocumentID)
if err != nil {
return fmt.Errorf("customschema: read max metadata version: %w", err)
}
nextVersion := maxVersion + 1
inserted, err := q.CreateDocumentCustomMetadata(txCtx, &repository.CreateDocumentCustomMetadataParams{
DocumentID: input.DocumentID,
Metadata: []byte(input.Metadata),
Version: nextVersion,
CreatedBy: actor,
})
if err != nil {
return fmt.Errorf("customschema: insert metadata row: %w", err)
}
newRow = inserted
return nil
})
}
// Belt-and-suspenders: retry once on (document_id, version) unique
// violation. The parent-row FOR UPDATE lock in step 1 serializes
// concurrent writers so this should never fire in practice, but a
// retry absorbs any edge case where two transactions somehow compute
// the same next_version before either commits.
txErr := runTx()
if txErr != nil && isUniqueVersionViolation(txErr) {
txErr = runTx()
}
if txErr != nil {
return nil, txErr
}
// Post-commit audit — the tx already released its locks, so a failure
// to persist an audit record is a non-fatal warning, but we still
// propagate it as an error for visibility (matches Milestone 1 style).
rec := AuditRecord{
Actor: actor,
Action: AuditActionMetadataWrite,
ResourceID: newRow.ID,
ClientID: schemaClientID,
Timestamp: time.Now().UTC(),
Details: map[string]any{
"document_id": newRow.DocumentID.String(),
"version": newRow.Version,
},
}
if auditErr := s.audit.Record(ctx, rec); auditErr != nil {
return nil, fmt.Errorf("customschema: record metadata audit: %w", auditErr)
}
return &CustomMetadata{
ID: newRow.ID,
DocumentID: newRow.DocumentID,
Metadata: append(json.RawMessage(nil), newRow.Metadata...),
Version: int(newRow.Version),
SchemaID: schemaID,
SchemaName: schemaName,
SchemaVersion: schemaVersion,
CreatedAt: newRow.CreatedAt.Time,
CreatedBy: newRow.CreatedBy,
}, nil
}
// GetCurrentMetadata returns the most-recent metadata row for a document,
// decorated with schema id/name/version via the joined query.
// ErrMetadataVersionNotFound is returned when the document has no rows.
// ErrDocumentNotFound is returned when the document itself is missing.
func (s *Service) GetCurrentMetadata(ctx context.Context, documentID uuid.UUID) (*CustomMetadata, error) {
if documentID == uuid.Nil {
return nil, fmt.Errorf("customschema: document id is required")
}
queries := s.cfg.GetDBQueries()
// Probe for the document existence first so a missing doc returns
// ErrDocumentNotFound rather than ErrMetadataVersionNotFound.
if _, err := queries.GetDocumentCustomSchemaId(ctx, documentID); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return nil, ErrDocumentNotFound
}
return nil, fmt.Errorf("customschema: probe document: %w", err)
}
row, err := queries.GetCurrentDocumentCustomMetadata(ctx, documentID)
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return nil, ErrMetadataVersionNotFound
}
return nil, fmt.Errorf("customschema: get current metadata: %w", err)
}
return currentRowToCustomMetadata(row), nil
}
// GetMetadataByVersion returns a specific historical version for a
// document. ErrMetadataVersionNotFound is returned when the version does
// not exist. ErrDocumentNotFound is returned when the document itself is
// missing.
func (s *Service) GetMetadataByVersion(ctx context.Context, documentID uuid.UUID, version int) (*CustomMetadata, error) {
if documentID == uuid.Nil {
return nil, fmt.Errorf("customschema: document id is required")
}
if version < 1 {
return nil, fmt.Errorf("customschema: version must be >= 1")
}
queries := s.cfg.GetDBQueries()
if _, err := queries.GetDocumentCustomSchemaId(ctx, documentID); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return nil, ErrDocumentNotFound
}
return nil, fmt.Errorf("customschema: probe document: %w", err)
}
row, err := queries.GetDocumentCustomMetadataByVersion(ctx, &repository.GetDocumentCustomMetadataByVersionParams{
DocumentID: documentID,
Version: int32(version), //nolint:gosec // metadata version is a per-doc monotonic counter, far below int32 ceiling
})
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return nil, ErrMetadataVersionNotFound
}
return nil, fmt.Errorf("customschema: get metadata by version: %w", err)
}
return byVersionRowToCustomMetadata(row), nil
}
// MaxMetadataHistoryLimit clamps the maximum page size for
// GetMetadataHistory. Callers requesting more than this are silently
// capped at the ceiling — the error path is reserved for obviously
// invalid values like 0 or negative numbers.
const MaxMetadataHistoryLimit = 200
// GetMetadataHistory returns a paged list of version summaries for a
// document in descending version order. limit must be in [1, 200]; offset
// must be >= 0. Out-of-range limit returns an error; excessive limit is
// clamped to MaxMetadataHistoryLimit.
func (s *Service) GetMetadataHistory(ctx context.Context, documentID uuid.UUID, limit, offset int) ([]MetadataVersion, error) {
if documentID == uuid.Nil {
return nil, fmt.Errorf("customschema: document id is required")
}
if limit <= 0 {
return nil, fmt.Errorf("customschema: limit must be between 1 and %d", MaxMetadataHistoryLimit)
}
if offset < 0 {
return nil, fmt.Errorf("customschema: offset must be >= 0")
}
if limit > MaxMetadataHistoryLimit {
limit = MaxMetadataHistoryLimit
}
queries := s.cfg.GetDBQueries()
if _, err := queries.GetDocumentCustomSchemaId(ctx, documentID); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return nil, ErrDocumentNotFound
}
return nil, fmt.Errorf("customschema: probe document: %w", err)
}
rows, err := queries.GetDocumentCustomMetadataHistory(ctx, &repository.GetDocumentCustomMetadataHistoryParams{
DocumentID: documentID,
LimitVal: int64(limit),
OffsetVal: int64(offset),
})
if err != nil {
return nil, fmt.Errorf("customschema: get metadata history: %w", err)
}
out := make([]MetadataVersion, 0, len(rows))
for _, r := range rows {
out = append(out, MetadataVersion{
Version: int(r.Version),
CreatedAt: r.CreatedAt.Time,
CreatedBy: r.CreatedBy,
})
}
return out, nil
}
// AssignSchema binds (or clears) a document's custom schema. The tx takes
// FOR UPDATE on the documents row as its FIRST statement so the
// mutual-exclusivity invariant holds under concurrent AssignSchema /
// CreateFieldExtraction calls. The rules enforced in order:
//
// 1. Document exists (pgx.ErrNoRows → ErrDocumentNotFound).
// 2. If schemaID is non-nil:
// a. Schema exists and is active.
// b. Schema belongs to the same client as the document (the composite
// FK enforces this at commit time; we convert the 23503 surface to a
// clean sentinel if it fires).
// 3. Document has no existing custom metadata (mirrors Trigger 1).
// 4. Document has no legacy field extractions (mirrors Trigger 1).
// 5. UPDATE documents SET custom_schema_id = ... WHERE id = ... .
// 6. Post-commit audit entry (schema.assign).
func (s *Service) AssignSchema(ctx context.Context, documentID uuid.UUID, schemaID *uuid.UUID, actor string) (*AssignSchemaResult, error) {
if documentID == uuid.Nil {
return nil, fmt.Errorf("customschema: document id is required")
}
var (
resultSchemaID *uuid.UUID
resultSchemaName *string
resultSchemaVersion *int
resultClientID string
)
txErr := s.cfg.ExecuteDBTransaction(ctx, func(txCtx context.Context, q *repository.Queries) error {
docClientID, err := s.assignSchemaLockAndCheck(txCtx, q, documentID, schemaID)
if err != nil {
return err
}
resultClientID = docClientID
if schemaID != nil {
schemaRow, err := s.assignSchemaValidateTarget(txCtx, q, *schemaID, docClientID)
if err != nil {
return err
}
idCopy := schemaRow.ID
resultSchemaID = &idCopy
nameCopy := schemaRow.Name
resultSchemaName = &nameCopy
verCopy := int(schemaRow.Version)
resultSchemaVersion = &verCopy
}
if err := q.SetDocumentCustomSchemaId(txCtx, &repository.SetDocumentCustomSchemaIdParams{
ID: documentID,
CustomSchemaID: schemaID,
}); err != nil {
// The composite FK can fire here if a race changed the
// clientId between the check above and the UPDATE. Map it
// to a clean sentinel.
var pgErr *pgconn.PgError
if errors.As(err, &pgErr) && pgErr.Code == "23503" &&
pgErr.ConstraintName == "fk_documents_custom_schema_same_client" {
return ErrSchemaClientMismatch
}
return fmt.Errorf("customschema: update document schema: %w", err)
}
return nil
})
if txErr != nil {
return nil, txErr
}
// Post-commit audit. Resource id is the document id (the thing being
// mutated); details carry the new schema id (or nil for clears).
details := map[string]any{}
if resultSchemaID != nil {
details["schema_id"] = resultSchemaID.String()
details["schema_name"] = *resultSchemaName
details["schema_version"] = *resultSchemaVersion
} else {
details["schema_id"] = nil
}
if err := s.audit.Record(ctx, AuditRecord{
Actor: actor,
Action: AuditActionSchemaAssign,
ResourceID: documentID,
ClientID: resultClientID,
Timestamp: time.Now().UTC(),
Details: details,
}); err != nil {
return nil, fmt.Errorf("customschema: record schema.assign audit: %w", err)
}
return &AssignSchemaResult{
DocumentID: documentID,
CustomSchemaID: resultSchemaID,
SchemaName: resultSchemaName,
SchemaVersion: resultSchemaVersion,
}, nil
}
// assignSchemaLockAndCheck runs the FOR UPDATE parent-row lock and the
// mutex invariant checks for AssignSchema. incomingSchemaID is the value
// being written; when it equals the document's current binding the call is
// idempotent and the mutation guards are skipped. Returns the document's
// client id on success.
func (s *Service) assignSchemaLockAndCheck(txCtx context.Context, q *repository.Queries, documentID uuid.UUID, incomingSchemaID *uuid.UUID) (string, error) {
lockRow, err := q.LockDocumentForLegacyExtractionWrite(txCtx, documentID)
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return "", ErrDocumentNotFound
}
return "", fmt.Errorf("customschema: lock document for assign schema: %w", err)
}
docClientID, err := q.GetDocumentClientID(txCtx, documentID)
if err != nil {
return "", fmt.Errorf("customschema: read document client id: %w", err)
}
// Idempotency: if the incoming binding already equals the current value,
// skip the mutation guards. The DB trigger treats an unchanged
// custom_schema_id UPDATE as a no-op, so the later UPDATE is safe.
if schemaIDsEqual(incomingSchemaID, lockRow.CustomSchemaID) {
return docClientID, nil
}
hasMetadata, err := q.HasDocumentCustomMetadata(txCtx, documentID)
if err != nil {
return "", fmt.Errorf("customschema: check existing metadata: %w", err)
}
if hasMetadata {
return "", ErrDocumentHasCustomMetadata
}
hasLegacy, err := q.HasFieldExtraction(txCtx, documentID)
if err != nil {
return "", fmt.Errorf("customschema: check legacy extractions: %w", err)
}
if hasLegacy {
return "", ErrDocumentHasLegacyExtractions
}
return docClientID, nil
}
// schemaIDsEqual returns true when both pointers represent the same schema
// binding: both nil (no schema) or both point to the same UUID.
func schemaIDsEqual(a, b *uuid.UUID) bool {
if a == nil && b == nil {
return true
}
if a == nil || b == nil {
return false
}
return *a == *b
}
// assignSchemaValidateTarget loads the target schema row, verifies it is
// active, and checks the cross-client invariant. Returns the schema row
// on success.
func (s *Service) assignSchemaValidateTarget(txCtx context.Context, q *repository.Queries, schemaID uuid.UUID, docClientID string) (*repository.ClientMetadataSchema, error) {
schemaRow, err := q.GetClientMetadataSchema(txCtx, schemaID)
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return nil, ErrSchemaNotFound
}
return nil, fmt.Errorf("customschema: load target schema: %w", err)
}
if schemaRow.Status != repository.SchemaStatusTypeActive {
return nil, ErrSchemaNotActive
}
if schemaRow.ClientID != docClientID {
return nil, ErrSchemaClientMismatch
}
return schemaRow, nil
}
// currentRowToCustomMetadata translates the joined row type into the
// service's CustomMetadata model.
func currentRowToCustomMetadata(row *repository.GetCurrentDocumentCustomMetadataRow) *CustomMetadata {
var schemaVersion *int
if row.SchemaVersion != nil {
v := int(*row.SchemaVersion)
schemaVersion = &v
}
return &CustomMetadata{
ID: row.ID,
DocumentID: row.DocumentID,
Metadata: append(json.RawMessage(nil), row.Metadata...),
Version: int(row.Version),
SchemaID: row.SchemaID,
SchemaName: row.SchemaName,
SchemaVersion: schemaVersion,
CreatedAt: row.CreatedAt.Time,
CreatedBy: row.CreatedBy,
}
}
// byVersionRowToCustomMetadata translates the version lookup row into the
// service's CustomMetadata model. Same shape as the current-version row
// type so the translation is near-identical.
func byVersionRowToCustomMetadata(row *repository.GetDocumentCustomMetadataByVersionRow) *CustomMetadata {
var schemaVersion *int
if row.SchemaVersion != nil {
v := int(*row.SchemaVersion)
schemaVersion = &v
}
return &CustomMetadata{
ID: row.ID,
DocumentID: row.DocumentID,
Metadata: append(json.RawMessage(nil), row.Metadata...),
Version: int(row.Version),
SchemaID: row.SchemaID,
SchemaName: row.SchemaName,
SchemaVersion: schemaVersion,
CreatedAt: row.CreatedAt.Time,
CreatedBy: row.CreatedBy,
}
}
// isUniqueVersionViolation returns true when err is a PostgreSQL 23505
// (unique_violation) on the uq_doc_custom_metadata_version constraint.
// Used by SetDocumentMetadata to decide whether a single retry is warranted.
func isUniqueVersionViolation(err error) bool {
var pgErr *pgconn.PgError
return errors.As(err, &pgErr) &&
pgErr.Code == "23505" &&
pgErr.ConstraintName == "uq_doc_custom_metadata_version"
}