silo/internal/migration/properties.go

// Package migration provides property schema migration functionality.
package migration

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"github.com/kindredsystems/silo/internal/db"
	"github.com/kindredsystems/silo/internal/schema"
	"github.com/rs/zerolog"
)

// PropertyMigrator handles property schema migrations.
type PropertyMigrator struct {
	db     *db.DB
	schema *schema.Schema
	logger zerolog.Logger
}

// NewPropertyMigrator creates a new property migrator.
func NewPropertyMigrator(database *db.DB, sch *schema.Schema, logger zerolog.Logger) *PropertyMigrator {
	return &PropertyMigrator{
		db:     database,
		schema: sch,
		logger: logger,
	}
}

// MigrationResult contains the results of a migration run.
type MigrationResult struct {
	MigrationID   string
	FromVersion   int
	ToVersion     int
	ItemsAffected int
	StartedAt     time.Time
	CompletedAt   time.Time
	Status        string
	Error         error
}

// MigrateToVersion migrates all items to the specified property schema version.
// This creates new revisions for items that have outdated property schemas.
func (m *PropertyMigrator) MigrateToVersion(ctx context.Context, targetVersion int) (*MigrationResult, error) {
	if m.schema.PropertySchemas == nil {
		return nil, fmt.Errorf("no property schema defined for %s", m.schema.Name)
	}

	result := &MigrationResult{
		FromVersion: 0, // Will be updated based on what we find
		ToVersion:   targetVersion,
		StartedAt:   time.Now(),
		Status:      "running",
	}

	// Record migration start
	var migrationID string
	err := m.db.Pool().QueryRow(ctx, `
		INSERT INTO property_migrations (schema_name, from_version, to_version, status)
		VALUES ($1, $2, $3, 'running')
		RETURNING id
	`, m.schema.Name, 0, targetVersion).Scan(&migrationID)
	if err != nil {
		return nil, fmt.Errorf("recording migration start: %w", err)
	}
	result.MigrationID = migrationID

	// Find items needing migration
	rows, err := m.db.Pool().Query(ctx, `
		SELECT DISTINCT i.id, i.part_number, r.properties, r.property_schema_version
		FROM items i
		JOIN revisions r ON r.item_id = i.id AND r.revision_number = i.current_revision
		WHERE i.archived_at IS NULL
		  AND (r.property_schema_version < $1 OR r.property_schema_version IS NULL)
	`, targetVersion)
	if err != nil {
		m.recordMigrationError(ctx, migrationID, err)
		return nil, fmt.Errorf("querying items: %w", err)
	}
	defer rows.Close()

	type itemToMigrate struct {
		ID         string
		PartNumber string
		Properties map[string]any
		Version    *int
	}

	var items []itemToMigrate
	for rows.Next() {
		var item itemToMigrate
		var propsJSON []byte
		if err := rows.Scan(&item.ID, &item.PartNumber, &propsJSON, &item.Version); err != nil {
			m.recordMigrationError(ctx, migrationID, err)
			return nil, fmt.Errorf("scanning row: %w", err)
		}
		if len(propsJSON) > 0 {
			if err := json.Unmarshal(propsJSON, &item.Properties); err != nil {
				m.logger.Warn().Err(err).Str("part_number", item.PartNumber).Msg("failed to parse properties")
				item.Properties = make(map[string]any)
			}
		} else {
			item.Properties = make(map[string]any)
		}
		items = append(items, item)
	}

	if err := rows.Err(); err != nil {
		m.recordMigrationError(ctx, migrationID, err)
		return nil, fmt.Errorf("iterating rows: %w", err)
	}

	m.logger.Info().Int("count", len(items)).Int("target_version", targetVersion).Msg("migrating items")

	// Migrate each item
	for _, item := range items {
		if err := m.migrateItem(ctx, item.ID, item.PartNumber, item.Properties, targetVersion); err != nil {
			m.logger.Error().Err(err).Str("part_number", item.PartNumber).Msg("failed to migrate item")
			// Continue with other items
		} else {
			result.ItemsAffected++
		}
	}

	// Record success
	result.CompletedAt = time.Now()
	result.Status = "completed"

	_, err = m.db.Pool().Exec(ctx, `
		UPDATE property_migrations
		SET completed_at = $1, status = 'completed', items_affected = $2
		WHERE id = $3
	`, result.CompletedAt, result.ItemsAffected, migrationID)
	if err != nil {
		m.logger.Warn().Err(err).Msg("failed to update migration record")
	}

	return result, nil
}

// migrateItem applies property defaults and creates a new revision.
func (m *PropertyMigrator) migrateItem(ctx context.Context, itemID, partNumber string, properties map[string]any, targetVersion int) error {
	// Get category from properties
	category, _ := properties["category"].(string)

	// Apply defaults for missing properties
	newProps := m.schema.PropertySchemas.ApplyDefaults(properties, category)

	propsJSON, err := json.Marshal(newProps)
	if err != nil {
		return fmt.Errorf("marshaling properties: %w", err)
	}

	// Create new revision with updated properties
	comment := fmt.Sprintf("Property schema migration to version %d", targetVersion)
	_, err = m.db.Pool().Exec(ctx, `
		INSERT INTO revisions (item_id, revision_number, properties, property_schema_version, comment)
		SELECT $1, current_revision + 1, $2, $3, $4
		FROM items WHERE id = $1
	`, itemID, propsJSON, targetVersion, comment)
	if err != nil {
		return fmt.Errorf("creating revision: %w", err)
	}

	m.logger.Debug().Str("part_number", partNumber).Int("version", targetVersion).Msg("migrated item")
	return nil
}

func (m *PropertyMigrator) recordMigrationError(ctx context.Context, migrationID string, err error) {
	_, dbErr := m.db.Pool().Exec(ctx, `
		UPDATE property_migrations
		SET completed_at = now(), status = 'failed', error_message = $1
		WHERE id = $2
	`, err.Error(), migrationID)
	if dbErr != nil {
		m.logger.Warn().Err(dbErr).Msg("failed to record migration error")
	}
}

// GetMigrationHistory returns recent property migrations.
func (m *PropertyMigrator) GetMigrationHistory(ctx context.Context, limit int) ([]MigrationResult, error) {
	rows, err := m.db.Pool().Query(ctx, `
		SELECT id, from_version, to_version, items_affected, started_at,
		       COALESCE(completed_at, now()), status, error_message
		FROM property_migrations
		WHERE schema_name = $1
		ORDER BY started_at DESC
		LIMIT $2
	`, m.schema.Name, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var results []MigrationResult
	for rows.Next() {
		var r MigrationResult
		var errMsg *string
		if err := rows.Scan(&r.MigrationID, &r.FromVersion, &r.ToVersion,
			&r.ItemsAffected, &r.StartedAt, &r.CompletedAt, &r.Status, &errMsg); err != nil {
			return nil, err
		}
		if errMsg != nil {
			r.Error = fmt.Errorf("%s", *errMsg)
		}
		results = append(results, r)
	}

	return results, rows.Err()
}