Automated Data Integrity Verification After Migration

Automated Data Integrity Checks After Migration

Data migration is always stressful. You've moved thousands of records, but are you sure nothing was lost? Relations intact? URLs not broken? Without automated checks, you're guessing. Manual spot-checking is a lottery: you verify 5% of records, but errors hide in the remaining 95%. Especially with nested comments, meta fields, or SEO tags — a single miss can cost traffic or functionality.

Recently we worked with a client: 10,000 posts, 50,000 comments migrating from WordPress to Laravel. Manual spot-checking (3 days, two engineers) found 30 broken links. Automated checks — in 1 day — uncovered 200 lost comments, 5% duplicate URLs, and 150 pages missing SEO titles. The difference is clear: automation is 3× faster and 20× more accurate.

We are a team of engineers with 5+ years of experience in complex migrations. We've developed a script set that gives a full integrity picture in 1–2 days. The scripts adapt to your CMS and database structure — whether it's PostgreSQL, MySQL, or MongoDB.

Why Automated Integrity Checks Are Essential

Manual spot-checking is inefficient. You risk losing 2–5% of records (especially nested comments or meta fields), breaking parent-child relations, creating duplicate URLs, and leaving SEO tags missing — all of which hurt ranking. Automation eliminates these risks. Compare the two approaches:

Typical consequences of missed errors:

What We Check: Detailed Checklist

Our engineers have implemented scripts for five key checks. Each comes with concrete metrics.

How Checksum Verification Works

We compare aggregated MD5 over critical fields. It detects even minor changes in data. Source: Wikipedia, MD5.

def checksum_check(source_db, target_db):
    """Compare checksums over critical fields"""
    # PostgreSQL
    source_hash = source_db.query_one("""
        SELECT md5(string_agg(
            md5(id::text || coalesce(email,'') || coalesce(slug,'')),
            ',' ORDER BY id
        )) as hash
        FROM articles
        WHERE status = 'published'
    """)
    target_hash = target_db.query_one("""
        SELECT md5(string_agg(
            md5(legacy_id || coalesce(email,'') || coalesce(slug,'')),
            ',' ORDER BY CAST(legacy_id AS INTEGER)
        )) as hash
        FROM articles
        WHERE status = 'published'
    """)
    return source_hash == target_hash

Record Count Verification

We compare record counts per content type, considering status filters (published/draft). Example code:

class MigrationValidator:
    def __init__(self, source_db, target_db):
        self.source = source_db
        self.target = target_db
        self.results = []

    def check_counts(self):
        tables = [
            ('posts', 'articles', "status='publish'", "status='published'"),
            ('users', 'users', None, None),
            ('comments', 'comments', "approved=1", "status='approved'"),
            ('categories', 'categories', None, None),
        ]
        for src_table, tgt_table, src_where, tgt_where in tables:
            src_count = self.source.count(src_table, src_where)
            tgt_count = self.target.count(tgt_table, tgt_where)
            status = 'OK' if src_count == tgt_count else 'MISMATCH'
            self.results.append({
                'check': f'count_{src_table}',
                'status': status,
                'source': src_count,
                'target': tgt_count,
                'diff': tgt_count - src_count
            })

Referential Integrity Check

We look for orphan records: articles without an author, comments without a parent article or parent comment.

def check_referential_integrity(target_db):
    issues = []

    # Articles without author
    orphaned_posts = target_db.query("""
        SELECT a.id, a.title
        FROM articles a
        LEFT JOIN users u ON a.author_id = u.id
        WHERE a.author_id IS NOT NULL AND u.id IS NULL
    """)
    if orphaned_posts:
        issues.append(f"Articles without valid author: {len(orphaned_posts)}")

    # Comments pointing to non-existent posts
    orphaned_comments = target_db.query("""
        SELECT c.id
        FROM comments c
        LEFT JOIN articles a ON c.post_id = a.id
        WHERE a.id IS NULL
    """)
    if orphaned_comments:
        issues.append(f"Orphaned comments: {len(orphaned_comments)}")

    # Child comments without parent
    broken_threads = target_db.query("""
        SELECT c.id
        FROM comments c
        LEFT JOIN comments p ON c.parent_id = p.id
        WHERE c.parent_id IS NOT NULL AND p.id IS NULL
    """)
    if broken_threads:
        issues.append(f"Comments with missing parent: {len(broken_threads)}")

    return issues

URL Availability Check

We asynchronously check every published URL for HTTP 200, 404, 500, and redirect chains.

import asyncio
import aiohttp

async def check_urls(urls, base_url, concurrency=20):
    errors = {'404': [], '500': [], 'redirect_chain': []}
    semaphore = asyncio.Semaphore(concurrency)

    async def check_one(session, path):
        async with semaphore:
            url = f"{base_url}{path}"
            try:
                async with session.get(url, allow_redirects=True) as resp:
                    if resp.status == 404:
                        errors['404'].append(path)
                    elif resp.status >= 500:
                        errors['500'].append(path)
                    elif len(resp.history) > 2:
                        errors['redirect_chain'].append(f"{path} ({len(resp.history)} redirects)")
            except Exception as e:
                errors['500'].append(f"{path} (error: {e})")

    async with aiohttp.ClientSession() as session:
        tasks = [check_one(session, url) for url in urls]
        await asyncio.gather(*tasks)
    return errors

# Run
urls_to_check = get_all_published_urls(target_db)
results = asyncio.run(check_urls(urls_to_check, 'https://new-site.com'))

SEO Metadata Check

We look for pages missing title/description, duplicate titles, and missing canonical tags.

def check_seo_completeness(target_db):
    issues = []

    # Pages without title
    no_title = target_db.query("""
        SELECT slug
        FROM articles
        WHERE (seo_title IS NULL OR seo_title = '')
          AND status = 'published'
    """)
    if no_title:
        issues.append(f"Pages without SEO title: {len(no_title)}")

    # Pages without meta description
    no_desc = target_db.query("""
        SELECT slug
        FROM articles
        WHERE (seo_description IS NULL OR seo_description = '')
          AND status = 'published'
    """)
    if no_desc:
        issues.append(f"Pages without meta description: {len(no_desc)}")

    # Duplicate titles
    dup_titles = target_db.query("""
        SELECT seo_title, COUNT(*) as count
        FROM articles
        WHERE status = 'published'
        GROUP BY seo_title
        HAVING COUNT(*) > 1
    """)
    if dup_titles:
        issues.append(f"Duplicate SEO titles: {len(dup_titles)} groups")

    return issues

Our Process

1. Schema Analysis — Study source and target database structures, identify critical tables and fields.
2. Script Adaptation — Tailor checks to your specific stack (PostgreSQL/MySQL, CMS).
3. Run & Collect Results — Execute scripts, generate report in HTML/JSON/JUnit.
4. Analysis & Recommendations — Break down each failed test, propose remediation plan.
5. CI/CD Deployment (optional) — Package checks into Docker container, connect to pipeline.

What's Included

• Check scripts (Python, configurable for your project).
• Detailed report with results table and list of problematic records.
• Remediation recommendations for each error.
• Documentation on running and interpreting results.
• Training for one client engineer.
• Guarantee: If after fixes a discrepancy is found, we refine the script at no extra cost.

Estimated Timeline

Script development and final report: 1–2 working days. Complexity depends on number of tables and data volume. Cost is calculated individually — contact us for an accurate estimate.

Why Choose Us

• 5+ years of migrations: from WordPress to custom Laravel solutions.
• 200+ data transfer projects.
• Certified engineers in PostgreSQL and Docker.
• Result guarantee: all checks run on a test environment before main execution.

Request a consultation on post-migration data integrity — we'll assess the scope and recommend the optimal set of checks. Contact us to get a detailed work plan and cost estimate.