Assistant/adapters/postgres/capability_repository.go

package postgres

import (
	"context"
	"errors"
	"fmt"

	"git.cin.su/m/Assistant/contracts/capability"
	"git.cin.su/m/Assistant/contracts/identity"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

var ErrCapabilityHealthNotFound = errors.New("capability health not found")
var ErrCapabilityBindingNotFound = errors.New("capability binding not found")
var ErrCapabilityBindingConflict = errors.New("capability binding conflict")
var ErrCapabilityImplementationNotFound = errors.New("capability implementation not found")

type CapabilityRepository struct{ pool *pgxpool.Pool }

func OpenCapability(ctx context.Context, config ContinuityConfig) (*CapabilityRepository, error) {
	pool, err := openPool(ctx, config)
	if err != nil {
		return nil, err
	}
	return &CapabilityRepository{pool: pool}, nil
}

func (r *CapabilityRepository) Close() { r.pool.Close() }

func (r *CapabilityRepository) AdmitCognitive(ctx context.Context, leaseID string, policy capability.AdmissionPolicy, request capability.AdmissionRequest, nowUnixMillis int64) (capability.AdmissionDecision, capability.AdmissionLease, error) {
	tx, err := r.pool.Begin(ctx)
	if err != nil {
		return capability.AdmissionDecision{}, capability.AdmissionLease{}, err
	}
	defer func() { _ = tx.Rollback(ctx) }()
	if leaseID == "" || nowUnixMillis < 1 {
		return capability.AdmissionDecision{State: capability.AdmissionRejected, Reason: capability.AdmissionReasonInputBudget, PolicyID: policy.ID}, capability.AdmissionLease{}, nil
	}
	var existingState string
	var existingPolicy string
	err = tx.QueryRow(ctx, `SELECT state, policy_id FROM capability.cognitive_admission_leases WHERE lease_id=$1`, leaseID).Scan(&existingState, &existingPolicy)
	if err == nil {
		if existingState == "ACTIVE" {
			return capability.AdmissionDecision{State: capability.AdmissionAdmitted, Reason: capability.AdmissionReasonNone, PolicyID: existingPolicy}, capability.AdmissionLease{ID: leaseID, PolicyID: existingPolicy}, tx.Commit(ctx)
		}
		return capability.AdmissionDecision{State: capability.AdmissionRejected, Reason: capability.AdmissionReasonConcurrency, PolicyID: existingPolicy}, capability.AdmissionLease{}, tx.Commit(ctx)
	}
	if !errors.Is(err, pgx.ErrNoRows) {
		return capability.AdmissionDecision{}, capability.AdmissionLease{}, err
	}
	windowStart := nowUnixMillis - 60*60*1000
	_, err = tx.Exec(ctx, `INSERT INTO capability.cognitive_admission_counters(policy_id,window_start_unix_millis,accepted_count,concurrent_count,updated_at_unix_millis) VALUES($1,$2,0,0,$3) ON CONFLICT(policy_id) DO NOTHING`, policy.ID, windowStart, nowUnixMillis)
	if err != nil {
		return capability.AdmissionDecision{}, capability.AdmissionLease{}, err
	}
	var storedWindow int64
	var accepted, concurrent int
	if err = tx.QueryRow(ctx, `SELECT window_start_unix_millis,accepted_count,concurrent_count FROM capability.cognitive_admission_counters WHERE policy_id=$1 FOR UPDATE`, policy.ID).Scan(&storedWindow, &accepted, &concurrent); err != nil {
		return capability.AdmissionDecision{}, capability.AdmissionLease{}, err
	}
	if nowUnixMillis-storedWindow >= 60*60*1000 {
		storedWindow, accepted = windowStart, 0
	}
	var expired int64
	if err = tx.QueryRow(ctx, `WITH expired AS (UPDATE capability.cognitive_admission_leases SET state='EXPIRED' WHERE policy_id=$1 AND state='ACTIVE' AND expires_at_unix_millis <= $2 RETURNING lease_id) SELECT count(*) FROM expired`, policy.ID, nowUnixMillis).Scan(&expired); err != nil {
		return capability.AdmissionDecision{}, capability.AdmissionLease{}, err
	}
	concurrent -= int(expired)
	if concurrent < 0 {
		concurrent = 0
	}
	decision := capability.Admit(request, policy, capability.AdmissionCounters{Concurrent: concurrent, AcceptedLastHour: accepted}, nowUnixMillis)
	if decision.State != capability.AdmissionAdmitted {
		if _, err = tx.Exec(ctx, `UPDATE capability.cognitive_admission_counters SET window_start_unix_millis=$2,accepted_count=$3,concurrent_count=$4,updated_at_unix_millis=$5 WHERE policy_id=$1`, policy.ID, storedWindow, accepted, concurrent, nowUnixMillis); err != nil {
			return capability.AdmissionDecision{}, capability.AdmissionLease{}, err
		}
		return decision, capability.AdmissionLease{}, tx.Commit(ctx)
	}
	expires := nowUnixMillis + 60*1000
	if request.DeadlineMillis > nowUnixMillis && request.DeadlineMillis < expires {
		expires = request.DeadlineMillis
	}
	if _, err = tx.Exec(ctx, `UPDATE capability.cognitive_admission_counters SET window_start_unix_millis=$2,accepted_count=$3,concurrent_count=$4,updated_at_unix_millis=$5 WHERE policy_id=$1`, policy.ID, storedWindow, accepted+1, concurrent+1, nowUnixMillis); err != nil {
		return capability.AdmissionDecision{}, capability.AdmissionLease{}, err
	}
	if _, err = tx.Exec(ctx, `INSERT INTO capability.cognitive_admission_leases(lease_id,policy_id,state,expires_at_unix_millis,created_at_unix_millis) VALUES($1,$2,'ACTIVE',$3,$4)`, leaseID, policy.ID, expires, nowUnixMillis); err != nil {
		return capability.AdmissionDecision{}, capability.AdmissionLease{}, err
	}
	return decision, capability.AdmissionLease{ID: leaseID, PolicyID: policy.ID}, tx.Commit(ctx)
}

func (r *CapabilityRepository) ReleaseCognitiveAdmission(ctx context.Context, lease capability.AdmissionLease, nowUnixMillis int64) error {
	tx, err := r.pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback(ctx) }()
	var policyID string
	err = tx.QueryRow(ctx, `UPDATE capability.cognitive_admission_leases SET state='RELEASED' WHERE lease_id=$1 AND state='ACTIVE' RETURNING policy_id`, lease.ID).Scan(&policyID)
	if errors.Is(err, pgx.ErrNoRows) {
		return tx.Commit(ctx)
	}
	if err != nil {
		return err
	}
	_, err = tx.Exec(ctx, `UPDATE capability.cognitive_admission_counters SET concurrent_count=GREATEST(0,concurrent_count-1),updated_at_unix_millis=$2 WHERE policy_id=$1`, policyID, nowUnixMillis)
	if err != nil {
		return err
	}
	return tx.Commit(ctx)
}

func (r *CapabilityRepository) SaveHealth(ctx context.Context, health capability.Health) error {
	_, err := r.pool.Exec(ctx, `INSERT INTO capability.health (capability_id, state, reason, revision, semantic_version, observed_at_unix_millis) VALUES ($1,$2,$3,$4,$5,$6) ON CONFLICT (capability_id) DO UPDATE SET state=EXCLUDED.state, reason=EXCLUDED.reason, revision=EXCLUDED.revision, semantic_version=EXCLUDED.semantic_version, observed_at_unix_millis=EXCLUDED.observed_at_unix_millis WHERE capability.health.revision < EXCLUDED.revision`, health.CapabilityID, health.State, health.Reason, health.Revision, health.SemanticVersion, health.ObservedAtUnixMillis)
	return err
}

func (r *CapabilityRepository) LoadHealth(ctx context.Context, id capability.CapabilityID) (capability.Health, error) {
	var health capability.Health
	err := r.pool.QueryRow(ctx, `SELECT capability_id, state, reason, revision, semantic_version, observed_at_unix_millis FROM capability.health WHERE capability_id=$1`, id).Scan(&health.CapabilityID, &health.State, &health.Reason, &health.Revision, &health.SemanticVersion, &health.ObservedAtUnixMillis)
	if err != nil {
		return capability.Health{}, fmt.Errorf("load capability health: %w", err)
	}
	return health, nil
}

func (r *CapabilityRepository) RebuildHealthProjection(ctx context.Context) (int64, error) {
	tx, err := r.pool.Begin(ctx)
	if err != nil {
		return 0, err
	}
	defer tx.Rollback(ctx)
	if _, err = tx.Exec(ctx, `DELETE FROM capability.health_projection`); err != nil {
		return 0, err
	}
	result, err := tx.Exec(ctx, `INSERT INTO capability.health_projection (capability_id, state, reason, revision, semantic_version, observed_at_unix_millis) SELECT capability_id, state, reason, revision, semantic_version, observed_at_unix_millis FROM capability.health`)
	if err != nil {
		return 0, err
	}
	if err = tx.Commit(ctx); err != nil {
		return 0, err
	}
	return result.RowsAffected(), nil
}

func (r *CapabilityRepository) SaveBinding(ctx context.Context, binding capability.Binding) error {
	tx, err := r.pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback(ctx) }()
	result, err := tx.Exec(ctx, `INSERT INTO capability.notification_bindings (
		binding_id,capability_id,owner_principal_id,revision,state,https_origin,delivery_path,receipt_path,
		protocol,secret_ref,verified_at_unix_millis,health)
		VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12) ON CONFLICT DO NOTHING`,
		binding.ID, binding.CapabilityID, binding.OwnerPrincipalID, binding.Revision, binding.State,
		binding.HTTPSOrigin, binding.DeliveryPath, binding.ReceiptPath, binding.Protocol, binding.SecretRef,
		binding.VerifiedAtUnixMillis, binding.Health)
	if err != nil {
		return err
	}
	if result.RowsAffected() == 0 {
		existing, loadErr := r.LoadBinding(ctx, binding.ID, binding.Revision)
		if loadErr != nil || existing != binding {
			return ErrCapabilityBindingConflict
		}
	}
	if binding.State == capability.BindingActive {
		result, err = tx.Exec(ctx, `INSERT INTO capability.notification_binding_heads (capability_id,owner_principal_id,binding_id,revision)
			VALUES ($1,$2,$3,$4) ON CONFLICT (capability_id,owner_principal_id) DO UPDATE SET binding_id=EXCLUDED.binding_id,revision=EXCLUDED.revision
			WHERE capability.notification_binding_heads.revision < EXCLUDED.revision`, binding.CapabilityID, binding.OwnerPrincipalID, binding.ID, binding.Revision)
		if err != nil || result.RowsAffected() != 1 {
			if err != nil {
				return err
			}
			return ErrCapabilityBindingConflict
		}
	} else {
		_, err = tx.Exec(ctx, `DELETE FROM capability.notification_binding_heads WHERE capability_id=$1 AND owner_principal_id=$2 AND binding_id=$3 AND revision < $4`, binding.CapabilityID, binding.OwnerPrincipalID, binding.ID, binding.Revision)
		if err != nil {
			return err
		}
		_, err = tx.Exec(ctx, `INSERT INTO capability.cognitive_binding_revision_heads (binding_id,revision) VALUES ($1,$2) ON CONFLICT DO NOTHING`, binding.ID, binding.Revision)
		if err != nil {
			return err
		}
	}
	return tx.Commit(ctx)
}

func (r *CapabilityRepository) SaveCapabilityImplementation(ctx context.Context, value capability.CapabilityImplementation) error {
	tx, err := r.pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback(ctx) }()
	_, err = tx.Exec(ctx, `INSERT INTO capability.capability_implementations (
		implementation_id,capability_id,capability_kind,owner_principal_id,revision,implementation_kind,target_ref,credential_ref,
		qualification,activation,health,authority_policy_ref,data_constraints,cost_score_millis,latency_score_millis,reliability_millis,observed_at_unix_millis)
		VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17) ON CONFLICT DO NOTHING`,
		value.ID, value.CapabilityID, value.CapabilityKind, value.OwnerPrincipalID, value.Revision, value.Kind, value.TargetRef,
		value.CredentialRef, value.Qualification, value.Activation, value.Health, value.AuthorityPolicyRef, value.DataConstraints,
		value.CostScoreMillis, value.LatencyScoreMillis, value.ReliabilityMillis, value.ObservedAtUnixMillis)
	if err != nil {
		return err
	}
	_, err = tx.Exec(ctx, `INSERT INTO capability.capability_implementation_heads(implementation_id,revision,capability_kind,owner_principal_id)
		VALUES ($1,$2,$3,$4)
		ON CONFLICT (implementation_id) DO UPDATE SET revision=EXCLUDED.revision,capability_kind=EXCLUDED.capability_kind,owner_principal_id=EXCLUDED.owner_principal_id,updated_at=CURRENT_TIMESTAMP
		WHERE capability.capability_implementation_heads.revision < EXCLUDED.revision`, value.ID, value.Revision, value.CapabilityKind, value.OwnerPrincipalID)
	if err != nil {
		return err
	}
	return tx.Commit(ctx)
}

func (r *CapabilityRepository) LoadCapabilityImplementations(ctx context.Context, kind capability.CapabilityKind, owner identity.PrincipalID) ([]capability.CapabilityImplementation, error) {
	rows, err := r.pool.Query(ctx, `SELECT
		i.implementation_id,i.capability_id,i.capability_kind,i.owner_principal_id,i.revision,i.implementation_kind,i.target_ref,COALESCE(i.credential_ref,''),
		i.qualification,i.activation,i.health,i.authority_policy_ref,i.data_constraints,i.cost_score_millis,i.latency_score_millis,i.reliability_millis,i.observed_at_unix_millis
		FROM capability.capability_implementation_heads h
		JOIN capability.capability_implementations i ON i.implementation_id=h.implementation_id AND i.revision=h.revision
		WHERE i.capability_kind=$1 AND i.owner_principal_id=$2
		ORDER BY i.implementation_id`, kind, owner)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	result := make([]capability.CapabilityImplementation, 0)
	for rows.Next() {
		var value capability.CapabilityImplementation
		if err = rows.Scan(&value.ID, &value.CapabilityID, &value.CapabilityKind, &value.OwnerPrincipalID, &value.Revision, &value.Kind, &value.TargetRef,
			&value.CredentialRef, &value.Qualification, &value.Activation, &value.Health, &value.AuthorityPolicyRef, &value.DataConstraints,
			&value.CostScoreMillis, &value.LatencyScoreMillis, &value.ReliabilityMillis, &value.ObservedAtUnixMillis); err != nil {
			return nil, err
		}
		result = append(result, value)
	}
	return result, rows.Err()
}

func (r *CapabilityRepository) AvailableCapabilityKinds(ctx context.Context, owner identity.PrincipalID) ([]capability.CapabilityKind, error) {
	rows, err := r.pool.Query(ctx, `SELECT DISTINCT i.capability_kind
		FROM capability.capability_implementation_heads h
		JOIN capability.capability_implementations i ON i.implementation_id=h.implementation_id AND i.revision=h.revision
		WHERE i.owner_principal_id=$1 AND i.qualification=$2 AND i.activation=$3 AND i.health=$4
		ORDER BY i.capability_kind`, owner, capability.QualificationValidated, capability.ActivationActive, capability.HealthAvailable)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	result := make([]capability.CapabilityKind, 0)
	for rows.Next() {
		var kind capability.CapabilityKind
		if err := rows.Scan(&kind); err != nil {
			return nil, err
		}
		result = append(result, kind)
	}
	return result, rows.Err()
}

func (r *CapabilityRepository) SaveCapabilityGap(ctx context.Context, value capability.CapabilityGap) error {
	strategies := make([]string, 0, len(value.Strategies))
	for _, strategy := range value.Strategies {
		strategies = append(strategies, string(strategy.Kind))
	}
	_, err := r.pool.Exec(ctx, `INSERT INTO capability.capability_gaps (
		gap_id,requirement_id,capability_kind,capability_id,owner_principal_id,original_intent_id,source_utterance_id,
		reason,detail,strategies,recorded_at_unix_millis)
		VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11) ON CONFLICT DO NOTHING`, value.ID, value.RequirementID,
		value.CapabilityKind, value.CapabilityID, value.OwnerPrincipalID, value.OriginalIntentID, value.SourceUtteranceID,
		value.Reason, value.Detail, strategies, value.RecordedAtUnixMillis)
	return err
}

func (r *CapabilityRepository) LatestCapabilityGap(ctx context.Context, kind capability.CapabilityKind, owner identity.PrincipalID) (capability.CapabilityGap, error) {
	var value capability.CapabilityGap
	var strategies []string
	err := r.pool.QueryRow(ctx, `SELECT gap_id,requirement_id,capability_kind,capability_id,owner_principal_id,original_intent_id,source_utterance_id,reason,detail,strategies,recorded_at_unix_millis
		FROM capability.capability_gaps WHERE capability_kind=$1 AND owner_principal_id=$2 ORDER BY recorded_at_unix_millis DESC, gap_id DESC LIMIT 1`, kind, owner).
		Scan(&value.ID, &value.RequirementID, &value.CapabilityKind, &value.CapabilityID, &value.OwnerPrincipalID, &value.OriginalIntentID, &value.SourceUtteranceID, &value.Reason, &value.Detail, &strategies, &value.RecordedAtUnixMillis)
	if err != nil {
		return capability.CapabilityGap{}, err
	}
	for _, strategy := range strategies {
		value.Strategies = append(value.Strategies, capability.AcquisitionStrategyCandidate{Kind: capability.AcquisitionStrategyKind(strategy)})
	}
	return value, nil
}

func (r *CapabilityRepository) SaveCapabilityUsageObservation(ctx context.Context, value capability.CapabilityUsageObservation) error {
	_, err := r.pool.Exec(ctx, `INSERT INTO capability.capability_usage_observations (
		observation_id,implementation_id,implementation_revision,capability_id,capability_kind,owner_principal_id,
		original_intent_id,source_utterance_id,outcome,outcome_ref,latency_millis,observed_at_unix_millis)
		VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12) ON CONFLICT DO NOTHING`, value.ID, value.ImplementationID,
		value.ImplementationRevision, value.CapabilityID, value.CapabilityKind, value.OwnerPrincipalID, value.OriginalIntentID,
		value.SourceUtteranceID, value.Outcome, value.OutcomeRef, value.LatencyMillis, value.ObservedAtUnixMillis)
	return err
}

func (r *CapabilityRepository) SaveSkillProfile(ctx context.Context, value capability.SkillProfile) error {
	tx, err := r.pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback(ctx) }()
	_, err = tx.Exec(ctx, `INSERT INTO capability.skill_profiles (
		skill_profile_id,revision,name,description,applicable_context,implementation_ref,implementation_version,status,
		supporting_experience_refs,known_failure_experience_refs,confidence_millis,last_validated_unix_millis,
		observed_failure_count,observed_correction_count,recorded_at_unix_millis)
		VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15) ON CONFLICT DO NOTHING`,
		value.ID, value.Revision, value.Name, value.Description, value.ApplicableContext, value.ImplementationRef,
		value.ImplementationVersion, value.Status, value.SupportingExperienceRefs, value.KnownFailureExperienceRefs,
		value.ConfidenceMillis, value.LastValidatedUnixMillis, value.ObservedFailureCount, value.ObservedCorrectionCount,
		value.RecordedAtUnixMillis)
	if err != nil {
		return err
	}
	_, err = tx.Exec(ctx, `INSERT INTO capability.skill_profile_heads(skill_profile_id,revision,status)
		VALUES ($1,$2,$3)
		ON CONFLICT (skill_profile_id) DO UPDATE SET revision=EXCLUDED.revision,status=EXCLUDED.status,updated_at=CURRENT_TIMESTAMP
		WHERE capability.skill_profile_heads.revision < EXCLUDED.revision`, value.ID, value.Revision, value.Status)
	if err != nil {
		return err
	}
	return tx.Commit(ctx)
}

func (r *CapabilityRepository) LoadSkillProfile(ctx context.Context, id capability.SkillProfileID) (capability.SkillProfile, error) {
	var value capability.SkillProfile
	err := r.pool.QueryRow(ctx, `SELECT
		p.skill_profile_id,p.revision,p.name,p.description,p.applicable_context,p.implementation_ref,p.implementation_version,p.status,
		p.supporting_experience_refs,p.known_failure_experience_refs,p.confidence_millis,p.last_validated_unix_millis,
		p.observed_failure_count,p.observed_correction_count,p.recorded_at_unix_millis
		FROM capability.skill_profile_heads h
		JOIN capability.skill_profiles p ON p.skill_profile_id=h.skill_profile_id AND p.revision=h.revision
		WHERE p.skill_profile_id=$1`, id).Scan(&value.ID, &value.Revision, &value.Name, &value.Description, &value.ApplicableContext,
		&value.ImplementationRef, &value.ImplementationVersion, &value.Status, &value.SupportingExperienceRefs,
		&value.KnownFailureExperienceRefs, &value.ConfidenceMillis, &value.LastValidatedUnixMillis,
		&value.ObservedFailureCount, &value.ObservedCorrectionCount, &value.RecordedAtUnixMillis)
	if err != nil {
		return capability.SkillProfile{}, fmt.Errorf("load skill profile: %w", err)
	}
	return value, nil
}

func (r *CapabilityRepository) RecentSkillProfiles(ctx context.Context, limit int) ([]capability.SkillProfile, error) {
	if limit < 1 || limit > 100 {
		return nil, fmt.Errorf("invalid skill profile limit")
	}
	rows, err := r.pool.Query(ctx, `SELECT
		p.skill_profile_id,p.revision,p.name,p.description,p.applicable_context,p.implementation_ref,p.implementation_version,p.status,
		p.supporting_experience_refs,p.known_failure_experience_refs,p.confidence_millis,p.last_validated_unix_millis,
		p.observed_failure_count,p.observed_correction_count,p.recorded_at_unix_millis
		FROM capability.skill_profile_heads h
		JOIN capability.skill_profiles p ON p.skill_profile_id=h.skill_profile_id AND p.revision=h.revision
		ORDER BY h.updated_at DESC, p.skill_profile_id LIMIT $1`, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var result []capability.SkillProfile
	for rows.Next() {
		var value capability.SkillProfile
		if err := rows.Scan(&value.ID, &value.Revision, &value.Name, &value.Description, &value.ApplicableContext,
			&value.ImplementationRef, &value.ImplementationVersion, &value.Status, &value.SupportingExperienceRefs,
			&value.KnownFailureExperienceRefs, &value.ConfidenceMillis, &value.LastValidatedUnixMillis,
			&value.ObservedFailureCount, &value.ObservedCorrectionCount, &value.RecordedAtUnixMillis); err != nil {
			return nil, err
		}
		result = append(result, value)
	}
	return result, rows.Err()
}

func (r *CapabilityRepository) LoadBinding(ctx context.Context, id capability.CapabilityBindingID, revision int64) (capability.Binding, error) {
	var binding capability.Binding
	err := r.pool.QueryRow(ctx, `SELECT binding_id,capability_id,owner_principal_id,revision,state,https_origin,delivery_path,receipt_path,
		protocol,secret_ref,verified_at_unix_millis,health FROM capability.notification_bindings WHERE binding_id=$1 AND revision=$2`, id, revision).
		Scan(&binding.ID, &binding.CapabilityID, &binding.OwnerPrincipalID, &binding.Revision, &binding.State,
			&binding.HTTPSOrigin, &binding.DeliveryPath, &binding.ReceiptPath, &binding.Protocol, &binding.SecretRef,
			&binding.VerifiedAtUnixMillis, &binding.Health)
	if err != nil {
		return capability.Binding{}, fmt.Errorf("load capability binding: %w", err)
	}
	return binding, nil
}

func (r *CapabilityRepository) LoadActiveBinding(ctx context.Context, capabilityID capability.CapabilityID, owner identity.PrincipalID) (capability.Binding, error) {
	var binding capability.Binding
	err := r.pool.QueryRow(ctx, `SELECT b.binding_id,b.capability_id,b.owner_principal_id,b.revision,b.state,b.https_origin,b.delivery_path,b.receipt_path,
		b.protocol,b.secret_ref,b.verified_at_unix_millis,b.health FROM capability.notification_binding_heads h
		JOIN capability.notification_bindings b ON b.binding_id=h.binding_id AND b.revision=h.revision
		WHERE h.capability_id=$1 AND h.owner_principal_id=$2 AND b.state=$3`, capabilityID, owner, capability.BindingActive).
		Scan(&binding.ID, &binding.CapabilityID, &binding.OwnerPrincipalID, &binding.Revision, &binding.State,
			&binding.HTTPSOrigin, &binding.DeliveryPath, &binding.ReceiptPath, &binding.Protocol, &binding.SecretRef,
			&binding.VerifiedAtUnixMillis, &binding.Health)
	if err != nil {
		return capability.Binding{}, fmt.Errorf("load active capability binding: %w", err)
	}
	return binding, nil
}

func (r *CapabilityRepository) SaveProviderInstance(ctx context.Context, value capability.ProviderInstance) error {
	_, err := r.pool.Exec(ctx, `INSERT INTO capability.cognitive_provider_instances
		(provider_instance_id,revision,provider_kind,endpoint_ref,credential_ref,enabled,qualification,health,observed_at_unix_millis,billing_mode,auth_mode,input_cost_micros,output_cost_micros,request_cost_micros)
		VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14) ON CONFLICT DO NOTHING`, value.ID, value.Revision, value.Kind,
		value.EndpointRef, value.CredentialRef, value.Enabled, value.Qualification, value.Health, value.ObservedAtUnixMillis, value.BillingMode, value.AuthMode, value.InputCostMicros, value.OutputCostMicros, value.RequestCostMicros)
	return err
}

func (r *CapabilityRepository) SaveModelDescriptor(ctx context.Context, value capability.ModelDescriptor) error {
	_, err := r.pool.Exec(ctx, `INSERT INTO capability.cognitive_model_descriptors
		(model_descriptor_id,revision,provider_instance_id,provider_model_id,fingerprint,max_context_bytes,modalities,capabilities,qualification,health,observed_at_unix_millis,intelligence_score_millis)
		VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12) ON CONFLICT DO NOTHING`, value.ID, value.Revision,
		value.ProviderInstanceID, value.ProviderModelID, value.Fingerprint, value.MaxContextBytes, value.Modalities,
		value.Capabilities, value.Qualification, value.Health, value.ObservedAtUnixMillis, value.IntelligenceScoreMillis)
	return err
}

func (r *CapabilityRepository) SaveCognitiveBinding(ctx context.Context, value capability.CognitiveBinding) error {
	_, err := r.pool.Exec(ctx, `INSERT INTO capability.cognitive_bindings
		(binding_id,revision,capability_id,owner_principal_id,provider_instance_id,provider_revision,model_descriptor_id,model_revision,data_policy,qualification,activation,health,limit_policy_ref)
		VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13) ON CONFLICT DO NOTHING`, value.ID, value.Revision,
		value.CapabilityID, value.OwnerPrincipalID, value.ProviderInstanceID, value.ProviderRevision, value.ModelDescriptorID,
		value.ModelRevision, value.DataPolicy, value.Qualification, value.Activation, value.Health, value.LimitPolicyRef)
	return err
}

func (r *CapabilityRepository) LoadCognitiveCandidates(ctx context.Context, capabilityID capability.CapabilityID, owner identity.PrincipalID) ([]capability.ResolvedCognitiveBinding, error) {
	rows, err := r.pool.Query(ctx, `SELECT
		b.binding_id,b.revision,b.capability_id,b.owner_principal_id,b.provider_instance_id,b.provider_revision,b.model_descriptor_id,b.model_revision,b.data_policy,b.qualification,b.activation,b.health,b.limit_policy_ref,
		p.provider_instance_id,p.revision,p.provider_kind,p.endpoint_ref,COALESCE(p.credential_ref,''),p.enabled,p.qualification,COALESCE(ph.health,p.health),GREATEST(p.observed_at_unix_millis,COALESCE(ph.observed_at_unix_millis,0)),p.billing_mode,p.auth_mode,p.input_cost_micros,p.output_cost_micros,p.request_cost_micros,
		m.model_descriptor_id,m.revision,m.provider_instance_id,m.provider_model_id,m.fingerprint,m.max_context_bytes,m.modalities,m.capabilities,m.qualification,COALESCE(ma.health,m.health),GREATEST(m.observed_at_unix_millis,COALESCE(ma.observed_at_unix_millis,0)),m.intelligence_score_millis
		FROM capability.cognitive_binding_revision_heads bh
		JOIN capability.cognitive_bindings b ON b.binding_id=bh.binding_id AND b.revision=bh.revision
		JOIN capability.cognitive_provider_instances p ON p.provider_instance_id=b.provider_instance_id AND p.revision=b.provider_revision
		JOIN capability.cognitive_model_descriptors m ON m.model_descriptor_id=b.model_descriptor_id AND m.revision=b.model_revision
		LEFT JOIN capability.cognitive_provider_health ph ON ph.provider_instance_id=p.provider_instance_id
		LEFT JOIN capability.cognitive_model_availability ma ON ma.provider_instance_id=p.provider_instance_id AND ma.model_descriptor_id=m.model_descriptor_id
		WHERE b.capability_id=$1 AND b.owner_principal_id=$2 ORDER BY b.binding_id`, capabilityID, owner)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var result []capability.ResolvedCognitiveBinding
	for rows.Next() {
		var value capability.ResolvedCognitiveBinding
		err = rows.Scan(&value.Binding.ID, &value.Binding.Revision, &value.Binding.CapabilityID, &value.Binding.OwnerPrincipalID, &value.Binding.ProviderInstanceID, &value.Binding.ProviderRevision, &value.Binding.ModelDescriptorID, &value.Binding.ModelRevision, &value.Binding.DataPolicy, &value.Binding.Qualification, &value.Binding.Activation, &value.Binding.Health, &value.Binding.LimitPolicyRef,
			&value.Provider.ID, &value.Provider.Revision, &value.Provider.Kind, &value.Provider.EndpointRef, &value.Provider.CredentialRef, &value.Provider.Enabled, &value.Provider.Qualification, &value.Provider.Health, &value.Provider.ObservedAtUnixMillis, &value.Provider.BillingMode, &value.Provider.AuthMode, &value.Provider.InputCostMicros, &value.Provider.OutputCostMicros, &value.Provider.RequestCostMicros,
			&value.Model.ID, &value.Model.Revision, &value.Model.ProviderInstanceID, &value.Model.ProviderModelID, &value.Model.Fingerprint, &value.Model.MaxContextBytes, &value.Model.Modalities, &value.Model.Capabilities, &value.Model.Qualification, &value.Model.Health, &value.Model.ObservedAtUnixMillis, &value.Model.IntelligenceScoreMillis)
		if err != nil {
			return nil, err
		}
		if err = r.loadCognitiveEvidence(ctx, &value); err != nil {
			return nil, err
		}
		result = append(result, value)
	}
	return result, rows.Err()
}

func (r *CapabilityRepository) loadCognitiveEvidence(ctx context.Context, value *capability.ResolvedCognitiveBinding) error {
	var evaluation capability.EvaluatedCapability
	err := r.pool.QueryRow(ctx, `SELECT evaluation_id,binding_id,binding_revision,capability_id,result,quality_score_millis,critical_failure_millis,benchmark_version,artifact_id,observed_at_unix_millis FROM capability.cognitive_evaluated_capabilities WHERE binding_id=$1 AND binding_revision=$2 ORDER BY observed_at_unix_millis DESC LIMIT 1`, value.Binding.ID, value.Binding.Revision).Scan(&evaluation.ID, &evaluation.BindingID, &evaluation.BindingRevision, &evaluation.CapabilityID, &evaluation.Result, &evaluation.QualityScoreMillis, &evaluation.CriticalFailureMillis, &evaluation.BenchmarkVersion, &evaluation.ArtifactID, &evaluation.ObservedAtUnixMillis)
	if err != nil && !errors.Is(err, pgx.ErrNoRows) {
		return err
	}
	if err == nil {
		value.Evaluation = &evaluation
	}
	var performance capability.PerformanceEvidence
	err = r.pool.QueryRow(ctx, `SELECT evidence_id,binding_id,binding_revision,model_descriptor_id,model_revision,hardware_profile,runtime,quantization,context_units,thinking_enabled,generation_tps_millis,prompt_tps_millis,ttft_micros,latency_micros,memory_bytes,reliability_millis,benchmark_version,artifact_id,observed_at_unix_millis FROM capability.cognitive_performance_evidence WHERE binding_id=$1 AND binding_revision=$2 ORDER BY observed_at_unix_millis DESC LIMIT 1`, value.Binding.ID, value.Binding.Revision).Scan(&performance.ID, &performance.BindingID, &performance.BindingRevision, &performance.ModelDescriptorID, &performance.ModelRevision, &performance.HardwareProfile, &performance.Runtime, &performance.Quantization, &performance.ContextUnits, &performance.ThinkingEnabled, &performance.GenerationTPSMillis, &performance.PromptTPSMillis, &performance.TTFTMicros, &performance.LatencyMicros, &performance.MemoryBytes, &performance.ReliabilityMillis, &performance.BenchmarkVersion, &performance.ArtifactID, &performance.ObservedAtUnixMillis)
	if err != nil && !errors.Is(err, pgx.ErrNoRows) {
		return err
	}
	if err == nil {
		value.Performance = &performance
	}
	return nil
}

func (r *CapabilityRepository) SaveCognitiveUsageObservation(ctx context.Context, value capability.CognitiveUsageObservation) error {
	_, err := r.pool.Exec(ctx, `INSERT INTO capability.cognitive_usage_observations
		(observation_id,request_id,binding_id,binding_revision,provider_instance_id,model_descriptor_id,observed_at_unix_millis,latency_millis,input_units,output_units,estimated_cost_micros,actual_cost_micros,result_class,failure_type)
		VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14) ON CONFLICT DO NOTHING`, value.ObservationID, value.RequestID,
		value.BindingID, value.BindingRevision, value.ProviderInstanceID, value.ModelDescriptorID, value.ObservedAtUnixMillis,
		value.LatencyMillis, value.InputUnits, value.OutputUnits, value.EstimatedCostMicros, value.ActualCostMicros, value.ResultClass, value.FailureType)
	return err
}

func (r *CapabilityRepository) SaveCognitiveDiscovery(ctx context.Context, value capability.CognitiveDiscoveryRegistration) error {
	tx, err := r.pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback(ctx) }()
	_, err = tx.Exec(ctx, `INSERT INTO capability.cognitive_provider_instances (provider_instance_id,revision,provider_kind,endpoint_ref,credential_ref,enabled,qualification,health,observed_at_unix_millis,billing_mode,auth_mode,input_cost_micros,output_cost_micros,request_cost_micros) VALUES ($1,$2,$3,$4,NULLIF($5,''),$6,$7,$8,$9,$10,$11,$12,$13,$14) ON CONFLICT (provider_instance_id,revision) DO UPDATE SET health=EXCLUDED.health,observed_at_unix_millis=EXCLUDED.observed_at_unix_millis,billing_mode=EXCLUDED.billing_mode,auth_mode=EXCLUDED.auth_mode,input_cost_micros=EXCLUDED.input_cost_micros,output_cost_micros=EXCLUDED.output_cost_micros,request_cost_micros=EXCLUDED.request_cost_micros WHERE capability.cognitive_provider_instances.observed_at_unix_millis <= EXCLUDED.observed_at_unix_millis`, value.Provider.ID, value.Provider.Revision, value.Provider.Kind, value.Provider.EndpointRef, value.Provider.CredentialRef, value.Provider.Enabled, value.Provider.Qualification, value.Provider.Health, value.Provider.ObservedAtUnixMillis, value.Provider.BillingMode, value.Provider.AuthMode, value.Provider.InputCostMicros, value.Provider.OutputCostMicros, value.Provider.RequestCostMicros)
	if err != nil {
		return err
	}
	_, err = tx.Exec(ctx, `INSERT INTO capability.cognitive_provider_health (provider_instance_id,health,observed_at_unix_millis,failure_class) VALUES ($1,$2,$3,$4) ON CONFLICT (provider_instance_id) DO UPDATE SET health=EXCLUDED.health,observed_at_unix_millis=EXCLUDED.observed_at_unix_millis,failure_class=EXCLUDED.failure_class WHERE capability.cognitive_provider_health.observed_at_unix_millis <= EXCLUDED.observed_at_unix_millis`, value.ProviderHealth.ProviderInstanceID, value.ProviderHealth.Health, value.ProviderHealth.ObservedAtUnixMillis, value.ProviderHealth.FailureClass)
	if err != nil {
		return err
	}
	for index, model := range value.Models {
		_, err = tx.Exec(ctx, `INSERT INTO capability.cognitive_model_descriptors (model_descriptor_id,revision,provider_instance_id,provider_model_id,fingerprint,max_context_bytes,modalities,capabilities,qualification,health,observed_at_unix_millis,intelligence_score_millis) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12) ON CONFLICT (model_descriptor_id,revision) DO UPDATE SET health=EXCLUDED.health,observed_at_unix_millis=EXCLUDED.observed_at_unix_millis,intelligence_score_millis=EXCLUDED.intelligence_score_millis WHERE capability.cognitive_model_descriptors.observed_at_unix_millis <= EXCLUDED.observed_at_unix_millis`, model.ID, model.Revision, model.ProviderInstanceID, model.ProviderModelID, model.Fingerprint, model.MaxContextBytes, model.Modalities, model.Capabilities, model.Qualification, model.Health, model.ObservedAtUnixMillis, model.IntelligenceScoreMillis)
		if err != nil {
			return err
		}
		availability := value.ModelAvailability[index]
		_, err = tx.Exec(ctx, `INSERT INTO capability.cognitive_model_availability (provider_instance_id,model_descriptor_id,installed,loaded,invokable,health,observed_at_unix_millis) VALUES ($1,$2,$3,$4,$5,$6,$7) ON CONFLICT (provider_instance_id,model_descriptor_id) DO UPDATE SET installed=EXCLUDED.installed,loaded=EXCLUDED.loaded,invokable=EXCLUDED.invokable,health=EXCLUDED.health,observed_at_unix_millis=EXCLUDED.observed_at_unix_millis WHERE capability.cognitive_model_availability.observed_at_unix_millis <= EXCLUDED.observed_at_unix_millis`, availability.ProviderInstanceID, availability.ModelDescriptorID, availability.Installed, availability.Loaded, availability.Invokable, availability.Health, availability.ObservedAtUnixMillis)
		if err != nil {
			return err
		}
		binding := value.Bindings[index]
		_, err = tx.Exec(ctx, `INSERT INTO capability.cognitive_bindings (binding_id,revision,capability_id,owner_principal_id,provider_instance_id,provider_revision,model_descriptor_id,model_revision,data_policy,qualification,activation,health,limit_policy_ref) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13) ON CONFLICT (binding_id,revision) DO UPDATE SET health=EXCLUDED.health WHERE capability.cognitive_bindings.health <> EXCLUDED.health`, binding.ID, binding.Revision, binding.CapabilityID, binding.OwnerPrincipalID, binding.ProviderInstanceID, binding.ProviderRevision, binding.ModelDescriptorID, binding.ModelRevision, binding.DataPolicy, binding.Qualification, binding.Activation, binding.Health, binding.LimitPolicyRef)
		if err != nil {
			return err
		}
		_, err = tx.Exec(ctx, `INSERT INTO capability.cognitive_binding_revision_heads (binding_id,revision) VALUES ($1,$2) ON CONFLICT DO NOTHING`, binding.ID, binding.Revision)
		if err != nil {
			return err
		}
	}
	return tx.Commit(ctx)
}

func (r *CapabilityRepository) SaveEvaluatedCapability(ctx context.Context, value capability.EvaluatedCapability) error {
	_, err := r.pool.Exec(ctx, `INSERT INTO capability.cognitive_evaluated_capabilities (evaluation_id,binding_id,binding_revision,capability_id,result,quality_score_millis,critical_failure_millis,benchmark_version,artifact_id,observed_at_unix_millis) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10) ON CONFLICT DO NOTHING`, value.ID, value.BindingID, value.BindingRevision, value.CapabilityID, value.Result, value.QualityScoreMillis, value.CriticalFailureMillis, value.BenchmarkVersion, value.ArtifactID, value.ObservedAtUnixMillis)
	return err
}

func (r *CapabilityRepository) SaveCognitiveCanaryPromotion(ctx context.Context, value capability.CognitiveCanaryPromotion) error {
	tx, err := r.pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback(ctx) }()
	p, m, b, e := value.Provider, value.Model, value.Binding, value.Evaluation
	if _, err = tx.Exec(ctx, `INSERT INTO capability.cognitive_provider_instances (provider_instance_id,revision,provider_kind,endpoint_ref,credential_ref,enabled,qualification,health,observed_at_unix_millis,billing_mode,auth_mode,input_cost_micros,output_cost_micros,request_cost_micros) VALUES ($1,$2,$3,$4,NULLIF($5,''),$6,$7,$8,$9,$10,$11,$12,$13,$14) ON CONFLICT DO NOTHING`, p.ID, p.Revision, p.Kind, p.EndpointRef, p.CredentialRef, p.Enabled, p.Qualification, p.Health, p.ObservedAtUnixMillis, p.BillingMode, p.AuthMode, p.InputCostMicros, p.OutputCostMicros, p.RequestCostMicros); err != nil {
		return err
	}
	if _, err = tx.Exec(ctx, `INSERT INTO capability.cognitive_model_descriptors (model_descriptor_id,revision,provider_instance_id,provider_model_id,fingerprint,max_context_bytes,modalities,capabilities,qualification,health,observed_at_unix_millis,intelligence_score_millis) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12) ON CONFLICT DO NOTHING`, m.ID, m.Revision, m.ProviderInstanceID, m.ProviderModelID, m.Fingerprint, m.MaxContextBytes, m.Modalities, m.Capabilities, m.Qualification, m.Health, m.ObservedAtUnixMillis, m.IntelligenceScoreMillis); err != nil {
		return err
	}
	if _, err = tx.Exec(ctx, `INSERT INTO capability.cognitive_bindings (binding_id,revision,capability_id,owner_principal_id,provider_instance_id,provider_revision,model_descriptor_id,model_revision,data_policy,qualification,activation,health,limit_policy_ref) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13) ON CONFLICT DO NOTHING`, b.ID, b.Revision, b.CapabilityID, b.OwnerPrincipalID, b.ProviderInstanceID, b.ProviderRevision, b.ModelDescriptorID, b.ModelRevision, b.DataPolicy, b.Qualification, b.Activation, b.Health, b.LimitPolicyRef); err != nil {
		return err
	}
	if _, err = tx.Exec(ctx, `INSERT INTO capability.cognitive_evaluated_capabilities (evaluation_id,binding_id,binding_revision,capability_id,result,quality_score_millis,critical_failure_millis,benchmark_version,artifact_id,observed_at_unix_millis) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10) ON CONFLICT DO NOTHING`, e.ID, e.BindingID, e.BindingRevision, e.CapabilityID, e.Result, e.QualityScoreMillis, e.CriticalFailureMillis, e.BenchmarkVersion, e.ArtifactID, e.ObservedAtUnixMillis); err != nil {
		return err
	}
	result, err := tx.Exec(ctx, `UPDATE capability.cognitive_binding_revision_heads SET revision=$1,updated_at=CURRENT_TIMESTAMP WHERE binding_id=$2 AND revision=$3`, b.Revision, b.ID, value.ExpectedBindingRevision)
	if err != nil {
		return err
	}
	if result.RowsAffected() != 1 {
		return ErrCapabilityBindingConflict
	}
	return tx.Commit(ctx)
}

func (r *CapabilityRepository) SaveCognitiveActivationPromotion(ctx context.Context, value capability.CognitiveActivationPromotion) error {
	tx, err := r.pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer func() { _ = tx.Rollback(ctx) }()
	b, e := value.Binding, value.Evaluation
	if _, err = tx.Exec(ctx, `INSERT INTO capability.cognitive_bindings (binding_id,revision,capability_id,owner_principal_id,provider_instance_id,provider_revision,model_descriptor_id,model_revision,data_policy,qualification,activation,health,limit_policy_ref) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13) ON CONFLICT DO NOTHING`, b.ID, b.Revision, b.CapabilityID, b.OwnerPrincipalID, b.ProviderInstanceID, b.ProviderRevision, b.ModelDescriptorID, b.ModelRevision, b.DataPolicy, b.Qualification, b.Activation, b.Health, b.LimitPolicyRef); err != nil {
		return err
	}
	if _, err = tx.Exec(ctx, `INSERT INTO capability.cognitive_evaluated_capabilities (evaluation_id,binding_id,binding_revision,capability_id,result,quality_score_millis,critical_failure_millis,benchmark_version,artifact_id,observed_at_unix_millis) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10) ON CONFLICT DO NOTHING`, e.ID, e.BindingID, e.BindingRevision, e.CapabilityID, e.Result, e.QualityScoreMillis, e.CriticalFailureMillis, e.BenchmarkVersion, e.ArtifactID, e.ObservedAtUnixMillis); err != nil {
		return err
	}
	if _, err = tx.Exec(ctx, `UPDATE capability.cognitive_bindings SET activation='DEPRECATED' WHERE capability_id=$1 AND owner_principal_id=$2 AND activation='ACTIVE' AND NOT (binding_id=$3 AND revision=$4)`, b.CapabilityID, b.OwnerPrincipalID, b.ID, b.Revision); err != nil {
		return err
	}
	result, err := tx.Exec(ctx, `UPDATE capability.cognitive_binding_revision_heads SET revision=$1,updated_at=CURRENT_TIMESTAMP WHERE binding_id=$2 AND revision=$3`, b.Revision, b.ID, value.ExpectedBindingRevision)
	if err != nil {
		return err
	}
	if result.RowsAffected() != 1 {
		return ErrCapabilityBindingConflict
	}
	return tx.Commit(ctx)
}

func (r *CapabilityRepository) SavePerformanceEvidence(ctx context.Context, value capability.PerformanceEvidence) error {
	_, err := r.pool.Exec(ctx, `INSERT INTO capability.cognitive_performance_evidence (evidence_id,binding_id,binding_revision,model_descriptor_id,model_revision,hardware_profile,runtime,quantization,context_units,thinking_enabled,generation_tps_millis,prompt_tps_millis,ttft_micros,latency_micros,memory_bytes,reliability_millis,benchmark_version,artifact_id,observed_at_unix_millis) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19) ON CONFLICT DO NOTHING`, value.ID, value.BindingID, value.BindingRevision, value.ModelDescriptorID, value.ModelRevision, value.HardwareProfile, value.Runtime, value.Quantization, value.ContextUnits, value.ThinkingEnabled, value.GenerationTPSMillis, value.PromptTPSMillis, value.TTFTMicros, value.LatencyMicros, value.MemoryBytes, value.ReliabilityMillis, value.BenchmarkVersion, value.ArtifactID, value.ObservedAtUnixMillis)
	return err
}