Function migrateMessagesToSessionItems

pkg/session/migrations.go:407–455

migrateMessagesToSessionItems migrates data from the messages JSON column to the session_items table

(ctx context.Context, db *sql.DB)

// migrateMessagesToSessionItems migrates data from the messages JSON column to the session_items table
func migrateMessagesToSessionItems(ctx context.Context, db *sql.DB) error {
	slog.InfoContext(ctx, "Starting migration of messages to session_items")

	// Get all sessions that have messages but no items yet
	rows, err := db.QueryContext(ctx, `
		SELECT s.id, s.messages 
		FROM sessions s 
		WHERE s.messages IS NOT NULL 
		  AND s.messages != '' 
		  AND s.messages != '[]'
		  AND NOT EXISTS (SELECT 1 FROM session_items si WHERE si.session_id = s.id)
	`)
	if err != nil {
		return fmt.Errorf("querying sessions: %w", err)
	}
	defer rows.Close()

	var sessionsToMigrate []struct {
		id       string
		messages string
	}

	for rows.Next() {
		var id, messages string
		if err := rows.Scan(&id, &messages); err != nil {
			return fmt.Errorf("scanning session: %w", err)
		}
		sessionsToMigrate = append(sessionsToMigrate, struct {
			id       string
			messages string
		}{id, messages})
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterating sessions: %w", err)
	}

	slog.InfoContext(ctx, "Found sessions to migrate", "count", len(sessionsToMigrate))

	// Migrate each session
	for _, sess := range sessionsToMigrate {
		if err := migrateSessionMessages(ctx, db, sess.id, sess.messages, ""); err != nil {
			slog.WarnContext(ctx, "Failed to migrate session, skipping", "session_id", sess.id, "error", err)
			continue
		}
	}

	slog.InfoContext(ctx, "Completed migration of messages to session_items")
	return nil
}

// migrateSessionMessages migrates a single session's messages to session_items
func migrateSessionMessages(ctx context.Context, db *sql.DB, sessionID, messagesJSON, parentID string) error {