DokuWiki Statistics Plugin: MySQL to SQLite Migration Script (no official migration path exists)

README.md

DokuWiki Statistics Plugin: MySQL to SQLite Migration

The DokuWiki Statistics Plugin by splitbrain switched its backend from MySQL to SQLite. There is no official migration path - updating the plugin creates an empty SQLite database and historical data is lost.

This script migrates all data from the old MySQL schema to the new SQLite schema.

The Challenge

The old MySQL schema is denormalized (stats_access contains session, browser, referrer, and pageview data in one row). The new SQLite schema is normalized across 14 tables with foreign keys. This migration handles the full decomposition.

Migration Phases (FK dependency order)

1. Phase 1 - Independent tables: iplocation, history, logins, users (aggregated from 3 tables)
2. Phase 2 - sessions (enriched with browser info from stats_access) + orphan stub sessions
3. Phase 3 - referers, pageviews, outlinks, search, edits, media
4. Phase 4 - searchwords (with old-to-new search ID remapping)

Usage

1. Backup first!

mysqldump --no-tablespaces -u <user> -p <database> \
  stats_access stats_session stats_iplocation stats_history stats_logins \
  stats_edits stats_media stats_outlinks stats_search stats_searchwords \
  stats_refseen stats_groups stats_lastseen > ~/statistics_mysql_backup.sql

cp /path/to/dokuwiki/data/meta/statistics.sqlite3 ~/statistics_sqlite_backup.sqlite3

2. Update the plugin

Update the statistics plugin to the latest version. Visit a wiki page to trigger SQLite database creation.

3. Configure

cp config.example.php config.php
# Edit config.php with your MySQL credentials and SQLite path

4. Dry run

php migrate.php --dry-run --verbose

5. Migrate

php migrate.php --verbose

6. Verify & fix permissions

# Check the output for "FK Check: PASSED" and matching row counts
# Then fix permissions:
sudo chown www-data:www-data /path/to/dokuwiki/data/meta/statistics.sqlite3
sudo chmod 664 /path/to/dokuwiki/data/meta/statistics.sqlite3

CLI Options

Option	Description
--dry-run	Preview migration without writing data
--verbose	Detailed progress output
--config=PATH	Custom config file path (default: ./config.php)
--phase=N	Run only phase N (1-4)
--skip-phase=N	Skip phase N

Known Issues & Workarounds

MySQL Collation Conflicts

Old MyISAM tables often use mixed collations (utf8_unicode_ci vs utf8_general_ci). The script handles this with explicit COLLATE clauses and SET collation_connection.

Large Number of Stub Sessions

Tables like edits, outlinks, and media may reference sessions not present in stats_session. The script creates stub sessions for these orphans. This is normal for older data from before session tracking was added.

Schema Version Differences

The iplocation table date column may be named dt or lastupd depending on the plugin version. The script auto-detects this.

SQLite File Permissions

The SQLite database, WAL, and SHM files must be writable by the user running the migration. If running as a different user than www-data, temporarily chmod 666 the files, then restore to 664 after migration.

Old stats_search Has No Session/IP

The old search table lacks session and ip columns. Migrated search entries use a placeholder session (__migration_placeholder__).

Tables NOT Migrated

Table	Reason
stats_lastseen	MEMORY engine, ephemeral data
stats_groups	Structurally incompatible; refilled automatically on login
campaigns (new)	No old equivalent

Requirements

• PHP 7.4+ with mysqli and pdo_sqlite extensions
• Read access to MySQL stats_* tables
• Write access to the SQLite database file

File Structure

migrate.php              -- Main CLI script
config.example.php       -- Example configuration
src/
  MysqlSource.php        -- MySQL connection and data extraction
  SqliteTarget.php       -- SQLite connection and data import
  RefererMapper.php      -- Referrer URL to ID mapping
  SessionResolver.php    -- Orphan session detection and stub creation

License

Public domain. Use at your own risk. Always backup before migrating.

config.example.php

<?php
/**
 * Configuration for DokuWiki Statistics MySQL → SQLite Migration
 *
 * Copy this file to config.php and adjust the values below.
 */

return [
    // MySQL connection settings
    'mysql' => [
        'host'     => 'localhost',
        'port'     => 3306,
        'database' => 'dokuwiki',
        'user'     => 'root',
        'password' => '',
        'prefix'   => 'stats_',  // Table prefix (default: stats_)
        'charset'  => 'utf8mb4',
    ],

    // SQLite target database path
    'sqlite' => [
        'path' => '/path/to/dokuwiki/data/meta/statistics.sqlite3',
    ],

    // Migration options
    'options' => [
        'batch_size'     => 5000,   // Rows per transaction for large tables
        'dry_run'        => false,  // Preview only, no writes
        'verbose'        => true,   // Detailed output
        'skip_phases'    => [],     // Array of phase numbers to skip (1-4)
        'clear_target'   => true,   // Clear target tables before import (for idempotent reruns)
    ],
];

migrate.php

#!/usr/bin/env php
<?php
/**
 * DokuWiki Statistics Plugin: MySQL → SQLite Migration
 *
 * Migrates historical statistics data from the old MySQL-based plugin
 * to the new SQLite-based schema (splitbrain/dokuwiki-plugin-statistics).
 *
 * Usage:
 *   php migrate.php [--dry-run] [--verbose] [--config=path/to/config.php]
 *   php migrate.php --help
 *
 * The old schema is denormalized (stats_access contains session, browser,
 * referrer, and pageview data in one row). The new schema normalizes this
 * across 14 tables with foreign keys.
 *
 * Migration phases (respecting FK dependencies):
 *   Phase 1: Independent tables (iplocation, history, logins, users)
 *   Phase 2: Sessions (depends on users)
 *   Phase 3: Data tables (pageviews, referers, outlinks, search, edits, media)
 *   Phase 4: Searchwords (depends on search)
 */

namespace DokuWikiMigration;

// ─── Bootstrap ───

if (php_sapi_name() !== 'cli') {
    die("This script must be run from the command line.\n");
}

require_once __DIR__ . '/src/MysqlSource.php';
require_once __DIR__ . '/src/SqliteTarget.php';
require_once __DIR__ . '/src/RefererMapper.php';
require_once __DIR__ . '/src/SessionResolver.php';

// ─── CLI argument parsing ───

$options = getopt('', ['dry-run', 'verbose', 'config:', 'help', 'skip-phase:', 'phase:']);

if (isset($options['help'])) {
    echo <<<HELP
DokuWiki Statistics Plugin: MySQL → SQLite Migration

Usage:
  php migrate.php [options]

Options:
  --config=PATH   Path to config.php (default: ./config.php)
  --dry-run       Preview migration without writing data
  --verbose       Show detailed progress output
  --phase=N       Run only phase N (1-4)
  --skip-phase=N  Skip phase N (can be specified multiple times)
  --help          Show this help

Phases:
  1  Independent tables (iplocation, history, logins, users)
  2  Sessions (depends on users)
  3  Data tables (pageviews, referers, outlinks, search, edits, media)
  4  Searchwords (depends on search)

HELP;
    exit(0);
}

$configPath = $options['config'] ?? __DIR__ . '/config.php';
if (!file_exists($configPath)) {
    fwrite(STDERR, "Config file not found: $configPath\n");
    fwrite(STDERR, "Copy config.example.php to config.php and adjust settings.\n");
    exit(1);
}

$config = require $configPath;
$dryRun = isset($options['dry-run']) || ($config['options']['dry_run'] ?? false);
$verbose = isset($options['verbose']) || ($config['options']['verbose'] ?? false);
$batchSize = $config['options']['batch_size'] ?? 5000;
$clearTarget = $config['options']['clear_target'] ?? true;

// Determine which phases to run
$onlyPhase = isset($options['phase']) ? (int)$options['phase'] : null;
$skipPhases = $config['options']['skip_phases'] ?? [];
if (isset($options['skip-phase'])) {
    $sp = $options['skip-phase'];
    $skipPhases = array_merge($skipPhases, is_array($sp) ? $sp : [$sp]);
}
$skipPhases = array_map('intval', $skipPhases);

// ─── Helper functions ───

function out(string $msg): void
{
    echo $msg . "\n";
}

function info(string $msg): void
{
    global $verbose;
    if ($verbose) {
        echo "  [INFO] $msg\n";
    }
}

function progress(string $label, int $current, int $total): void
{
    global $verbose;
    if (!$verbose) return;
    $pct = $total > 0 ? round($current / $total * 100, 1) : 100;
    echo "  [$label] $current / $total ($pct%)\r";
    if ($current >= $total) echo "\n";
}

function shouldRunPhase(int $phase): bool
{
    global $onlyPhase, $skipPhases;
    if ($onlyPhase !== null) return $phase === $onlyPhase;
    return !in_array($phase, $skipPhases, true);
}

function formatDuration(float $seconds): string
{
    if ($seconds < 60) return round($seconds, 1) . 's';
    $min = floor($seconds / 60);
    $sec = round($seconds - $min * 60, 1);
    return "{$min}m {$sec}s";
}

// ─── Start migration ───

$totalStart = microtime(true);

out("╔══════════════════════════════════════════════════════════╗");
out("║  DokuWiki Statistics: MySQL → SQLite Migration          ║");
out("╚══════════════════════════════════════════════════════════╝");
out("");

if ($dryRun) {
    out("  *** DRY RUN MODE - No data will be written ***");
    out("");
}

// Connect to databases
out("[1/6] Connecting to databases...");
try {
    $mysql = new MysqlSource($config['mysql'], $batchSize);
    out("  MySQL: Connected to {$config['mysql']['host']}:{$config['mysql']['database']}");
} catch (\Exception $e) {
    fwrite(STDERR, "MySQL connection failed: " . $e->getMessage() . "\n");
    exit(1);
}

try {
    $sqlite = new SqliteTarget($config['sqlite']['path'], $batchSize, $dryRun);
    out("  SQLite: Opened {$config['sqlite']['path']}");
} catch (\Exception $e) {
    fwrite(STDERR, "SQLite connection failed: " . $e->getMessage() . "\n");
    $mysql->close();
    exit(1);
}

$refMapper = new RefererMapper($sqlite);
$sessionResolver = new SessionResolver($mysql, $sqlite);

// ─── Source counts ───

out("");
out("[2/6] Reading source table counts...");
$sourceCounts = [];
$sourceTableNames = [
    'iplocation', 'history', 'logins', 'session', 'access',
    'outlinks', 'search', 'searchwords', 'edits', 'media', 'refseen',
];
foreach ($sourceTableNames as $t) {
    try {
        $sourceCounts[$t] = $mysql->count($t);
        info("stats_$t: " . number_format($sourceCounts[$t]) . " rows");
    } catch (\Exception $e) {
        $sourceCounts[$t] = -1;
        info("stats_$t: ERROR - " . $e->getMessage());
    }
}
out("  Total source rows: " . number_format(array_sum(array_filter($sourceCounts, fn($v) => $v > 0))));
out("");

// ═══════════════════════════════════════════════════════════
// Phase 1: Independent tables
// ═══════════════════════════════════════════════════════════

if (shouldRunPhase(1)) {
    out("[3/6] Phase 1: Independent tables (iplocation, history, logins, users)");
    $phaseStart = microtime(true);

    // --- iplocation ---
    $count = $sourceCounts['iplocation'];
    out("  Migrating iplocation ($count rows)...");
    if ($clearTarget) $sqlite->clearTable('iplocation');
    $migrated = 0;
    foreach ($mysql->readIplocation() as $batch) {
        $sqlite->beginTransaction();
        foreach ($batch as $row) {
            $sqlite->insertIplocation($row);
            $migrated++;
        }
        $sqlite->commit();
        progress('iplocation', $migrated, $count);
    }
    out("  iplocation: $migrated rows migrated");

    // --- history ---
    $count = $sourceCounts['history'];
    out("  Migrating history ($count rows)...");
    if ($clearTarget) $sqlite->clearTable('history');
    $migrated = 0;
    foreach ($mysql->readHistory() as $batch) {
        $sqlite->beginTransaction();
        foreach ($batch as $row) {
            $sqlite->insertHistory($row);
            $migrated++;
        }
        $sqlite->commit();
        progress('history', $migrated, $count);
    }
    out("  history: $migrated rows migrated");

    // --- logins ---
    $count = $sourceCounts['logins'];
    out("  Migrating logins ($count rows)...");
    if ($clearTarget) $sqlite->clearTable('logins');
    $migrated = 0;
    foreach ($mysql->readLogins() as $batch) {
        $sqlite->beginTransaction();
        foreach ($batch as $row) {
            $sqlite->insertLogin($row);
            $migrated++;
        }
        $sqlite->commit();
        progress('logins', $migrated, $count);
    }
    out("  logins: $migrated rows migrated");

    // --- users (aggregated) ---
    out("  Aggregating users from access/edits/logins...");
    if ($clearTarget) $sqlite->clearTable('users');
    $users = $mysql->readDistinctUsers();
    $sqlite->beginTransaction();
    foreach ($users as $u) {
        // Attempt to extract domain from user if it contains @
        $domain = null;
        if (strpos($u['user'], '@') !== false) {
            $domain = substr($u['user'], strrpos($u['user'], '@') + 1);
        }
        $sqlite->insertUser($u['user'], $u['first_seen'], $domain);
    }
    $sqlite->commit();
    out("  users: " . count($users) . " distinct users migrated");

    $phaseDuration = microtime(true) - $phaseStart;
    out("  Phase 1 completed in " . formatDuration($phaseDuration));
    out("");
} else {
    out("[3/6] Phase 1: SKIPPED");
    out("");
}

// ═══════════════════════════════════════════════════════════
// Phase 2: Sessions (depends on users)
// ═══════════════════════════════════════════════════════════

if (shouldRunPhase(2)) {
    out("[4/6] Phase 2: Sessions");
    $phaseStart = microtime(true);

    $count = $sourceCounts['session'];
    out("  Building browser info map from stats_access...");
    $browserMap = $mysql->buildBrowserInfoMap();
    info("Browser info loaded for " . count($browserMap) . " sessions");

    out("  Migrating sessions ($count sessions)...");
    if ($clearTarget) $sqlite->clearTable('sessions');
    $migrated = 0;

    foreach ($mysql->readSessionsSimple() as $batch) {
        $sqlite->beginTransaction();
        foreach ($batch as $row) {
            $sid = $row['session'];
            $bi = $browserMap[$sid] ?? [];
            $sqlite->insertSession([
                'session' => $sid,
                'dt'      => $row['dt'],
                'end'     => $row['end'] ?? $row['dt'],
                'uid'     => $row['uid'] ?? '',
                'user'    => !empty($bi['user']) ? $bi['user'] : null,
                'ua'      => $bi['ua'] ?? '',
                'ua_info' => $bi['ua_info'] ?? '',
                'ua_type' => $bi['ua_type'] ?? 'browser',
                'ua_ver'  => $bi['ua_ver'] ?? '',
                'os'      => $bi['os'] ?? '',
            ]);
            $sessionResolver->markKnown($sid);
            $migrated++;
        }
        $sqlite->commit();
        progress('sessions', $migrated, $count);
    }
    // Pass browser map to session resolver for orphan enrichment
    $sessionResolver->setBrowserMap($browserMap);
    unset($browserMap); // Free the local reference (resolver holds it now)
    out("  sessions: $migrated rows migrated");

    // Resolve orphan sessions (uses browserMap, no per-session MySQL queries)
    out("  Resolving orphan sessions...");
    $sqlite->beginTransaction();
    $stubs = $sessionResolver->resolveOrphans();
    $sqlite->commit();
    out("  sessions: $stubs stub sessions created for orphans");

    // Create placeholder session for search entries
    $sqlite->beginTransaction();
    $sessionResolver->ensurePlaceholder();
    $sqlite->commit();
    out("  sessions: placeholder session created for search migration");

    $phaseDuration = microtime(true) - $phaseStart;
    out("  Phase 2 completed in " . formatDuration($phaseDuration));
    out("");
} else {
    out("[4/6] Phase 2: SKIPPED");
    out("");
}

// ═══════════════════════════════════════════════════════════
// Phase 3: Data tables (depends on sessions, referers)
// ═══════════════════════════════════════════════════════════

if (shouldRunPhase(3)) {
    out("[5/6] Phase 3: Data tables (referers, pageviews, outlinks, search, edits, media)");
    $phaseStart = microtime(true);

    // --- referers (build map first) ---
    out("  Building referrer map from stats_access...");
    $referers = $mysql->readDistinctReferers();
    $sqlite->beginTransaction();
    $refMapper->buildFromMysqlData($referers);
    $sqlite->commit();
    out("  referers: " . $refMapper->getMappedCount() . " distinct referrers mapped");

    // --- pageviews (largest table) ---
    $count = $sourceCounts['access'];
    out("  Migrating pageviews from stats_access ($count rows)...");
    if ($clearTarget) $sqlite->clearTable('pageviews');
    $migrated = 0;

    foreach ($mysql->readAccess() as $batch) {
        $sqlite->beginTransaction();
        foreach ($batch as $row) {
            // Resolve session
            $sessionId = $sessionResolver->ensure($row['session'] ?? '');

            // Resolve referrer
            $refId = null;
            $refType = $row['ref_type'] ?? '';
            $refUrl = $row['ref'] ?? '';
            if (!empty($refUrl) && in_array($refType, ['external', 'search'], true)) {
                $refId = $refMapper->resolve($refUrl, $refType);
            }

            $sqlite->insertPageview([
                'dt'       => $row['dt'],
                'ip'       => $row['ip'] ?? '',
                'session'  => $sessionId,
                'page'     => $row['page'] ?? '',
                'ref_id'   => $refId,
                'screen_x' => $row['screen_x'] ?? 0,
                'screen_y' => $row['screen_y'] ?? 0,
                'view_x'   => $row['view_x'] ?? 0,
                'view_y'   => $row['view_y'] ?? 0,
            ]);
            $migrated++;
        }
        $sqlite->commit();
        progress('pageviews', $migrated, $count);
    }
    out("  pageviews: $migrated rows migrated");

    // --- outlinks ---
    $count = $sourceCounts['outlinks'];
    out("  Migrating outlinks ($count rows)...");
    if ($clearTarget) $sqlite->clearTable('outlinks');
    $migrated = 0;
    foreach ($mysql->readOutlinks() as $batch) {
        $sqlite->beginTransaction();
        foreach ($batch as $row) {
            $sessionId = $sessionResolver->ensure($row['session'] ?? '');
            $sqlite->insertOutlink([
                'dt'      => $row['dt'],
                'session' => $sessionId,
                'link'    => $row['link'] ?? '',
                'page'    => $row['page'] ?? '',
            ]);
            $migrated++;
        }
        $sqlite->commit();
        progress('outlinks', $migrated, $count);
    }
    out("  outlinks: $migrated rows migrated");

    // --- search ---
    // Old stats_search has no session/ip columns. Use placeholder session.
    $count = $sourceCounts['search'];
    out("  Migrating search ($count rows)...");
    if ($clearTarget) {
        $sqlite->clearTable('searchwords'); // Clear dependent table first
        $sqlite->clearTable('search');
    }
    $migrated = 0;

    // Build old→new ID map for searchwords phase
    $searchIdMap = []; // old MySQL id → new SQLite id

    foreach ($mysql->readSearch() as $batch) {
        $sqlite->beginTransaction();
        foreach ($batch as $row) {
            $oldId = (int)$row['id'];
            $newId = $sqlite->insertSearch([
                'dt'      => $row['dt'],
                'ip'      => '',  // Not available in old schema
                'session' => SessionResolver::PLACEHOLDER_SESSION,
                'query'   => $row['query'] ?? '',
            ]);
            $searchIdMap[$oldId] = $newId;
            $migrated++;
        }
        $sqlite->commit();
        progress('search', $migrated, $count);
    }
    out("  search: $migrated rows migrated (with placeholder session)");

    // --- edits ---
    $count = $sourceCounts['edits'];
    out("  Migrating edits ($count rows)...");
    if ($clearTarget) $sqlite->clearTable('edits');
    $migrated = 0;
    foreach ($mysql->readEdits() as $batch) {
        $sqlite->beginTransaction();
        foreach ($batch as $row) {
            $sessionId = $sessionResolver->ensure($row['session'] ?? '');
            $sqlite->insertEdit([
                'dt'      => $row['dt'],
                'ip'      => $row['ip'] ?? '',
                'session' => $sessionId,
                'page'    => $row['page'] ?? '',
                'type'    => $row['type'] ?? '',
            ]);
            $migrated++;
        }
        $sqlite->commit();
        progress('edits', $migrated, $count);
    }
    out("  edits: $migrated rows migrated");

    // --- media ---
    $count = $sourceCounts['media'];
    out("  Migrating media ($count rows)...");
    if ($clearTarget) $sqlite->clearTable('media');
    $migrated = 0;
    foreach ($mysql->readMedia() as $batch) {
        $sqlite->beginTransaction();
        foreach ($batch as $row) {
            $sessionId = $sessionResolver->ensure($row['session'] ?? '');
            $sqlite->insertMedia([
                'dt'      => $row['dt'],
                'ip'      => $row['ip'] ?? '',
                'session' => $sessionId,
                'media'   => $row['media'] ?? '',
                'size'    => $row['size'] ?? 0,
                'mime1'   => $row['mime1'] ?? '',
                'mime2'   => $row['mime2'] ?? '',
                'inline'  => $row['inline'] ?? 0,
            ]);
            $migrated++;
        }
        $sqlite->commit();
        progress('media', $migrated, $count);
    }
    out("  media: $migrated rows migrated");

    $phaseDuration = microtime(true) - $phaseStart;
    out("  Phase 3 completed in " . formatDuration($phaseDuration));
    out("");
} else {
    out("[5/6] Phase 3: SKIPPED");
    $searchIdMap = [];
    out("");
}

// ═══════════════════════════════════════════════════════════
// Phase 4: Searchwords (depends on search)
// ═══════════════════════════════════════════════════════════

if (shouldRunPhase(4)) {
    out("[6/6] Phase 4: Searchwords");
    $phaseStart = microtime(true);

    $count = $sourceCounts['searchwords'];
    out("  Migrating searchwords ($count rows)...");
    if (empty($searchIdMap) && !shouldRunPhase(3)) {
        out("  WARNING: Phase 3 was skipped - searchwords need the search ID map.");
        out("  Building search ID map from existing data...");
        // Rebuild map: read old search IDs and match by query+dt
        out("  SKIPPED: Cannot map searchwords without Phase 3 data.");
    } else {
        if ($clearTarget && shouldRunPhase(3)) {
            // Already cleared in phase 3
        } elseif ($clearTarget) {
            $sqlite->clearTable('searchwords');
        }

        $migrated = 0;
        $skipped = 0;
        foreach ($mysql->readSearchwords() as $batch) {
            $sqlite->beginTransaction();
            foreach ($batch as $row) {
                $oldSid = (int)$row['sid'];
                if (!isset($searchIdMap[$oldSid])) {
                    $skipped++;
                    continue;
                }
                $newSid = $searchIdMap[$oldSid];
                $sqlite->insertSearchword($newSid, $row['word']);
                $migrated++;
            }
            $sqlite->commit();
            progress('searchwords', $migrated + $skipped, $count);
        }
        out("  searchwords: $migrated rows migrated, $skipped skipped (orphan sid)");
    }

    $phaseDuration = microtime(true) - $phaseStart;
    out("  Phase 4 completed in " . formatDuration($phaseDuration));
    out("");
} else {
    out("[6/6] Phase 4: SKIPPED");
    out("");
}

// ═══════════════════════════════════════════════════════════
// Verification
// ═══════════════════════════════════════════════════════════

out("═══════════════════════════════════════════════════════════");
out("Verification");
out("═══════════════════════════════════════════════════════════");

if (!$dryRun) {
    // Row counts
    $targetTables = [
        'iplocation'  => 'iplocation',
        'history'     => 'history',
        'logins'      => 'logins',
        'users'       => null, // aggregated, no 1:1 source
        'sessions'    => 'session',
        'referers'    => null, // aggregated
        'pageviews'   => 'access',
        'outlinks'    => 'outlinks',
        'search'      => 'search',
        'searchwords' => 'searchwords',
        'edits'       => 'edits',
        'media'       => 'media',
    ];

    out("");
    out("  Target Table     | Count       | Source Count | Match");
    out("  -----------------+-------------+-------------+------");

    foreach ($targetTables as $target => $sourceKey) {
        $targetCount = $sqlite->count($target);
        $sourceCount = $sourceKey !== null ? ($sourceCounts[$sourceKey] ?? '?') : 'N/A';

        $match = '';
        if (is_int($sourceCount)) {
            if ($target === 'sessions') {
                // Sessions may have stubs, so target >= source
                $match = $targetCount >= $sourceCount ? 'OK' : 'WARN';
            } else {
                $match = $targetCount === $sourceCount ? 'OK' : 'DIFF';
            }
        }

        $targetFormatted = str_pad(number_format($targetCount), 11, ' ', STR_PAD_LEFT);
        $sourceFormatted = is_int($sourceCount)
            ? str_pad(number_format($sourceCount), 11, ' ', STR_PAD_LEFT)
            : str_pad((string)$sourceCount, 11, ' ', STR_PAD_LEFT);

        out(sprintf("  %-17s| %s | %s | %s", $target, $targetFormatted, $sourceFormatted, $match));
    }

    // FK integrity check
    out("");
    out("  Foreign key integrity check...");
    $fkViolations = $sqlite->checkForeignKeys();
    if (empty($fkViolations)) {
        out("  FK Check: PASSED (no violations)");
    } else {
        out("  FK Check: FAILED (" . count($fkViolations) . " violations)");
        foreach (array_slice($fkViolations, 0, 10) as $v) {
            out("    Table: {$v['table']}, Row: {$v['rowid']}, Parent: {$v['parent']}, FK: {$v['fkid']}");
        }
        if (count($fkViolations) > 10) {
            out("    ... and " . (count($fkViolations) - 10) . " more");
        }
    }

    out("");
    out("  Session resolver stats:");
    out("    Known sessions: " . number_format($sessionResolver->getKnownCount()));
    out("    Stub sessions created: " . number_format($sessionResolver->getStubsCreated()));
    out("    Referrers mapped: " . number_format($refMapper->getMappedCount()));
} else {
    out("  (Skipped in dry-run mode)");
}

// ─── Cleanup ───

$sqlite->close();
$mysql->close();

$totalDuration = microtime(true) - $totalStart;
out("");
out("═══════════════════════════════════════════════════════════");
if ($dryRun) {
    out("  DRY RUN completed in " . formatDuration($totalDuration));
} else {
    out("  Migration completed in " . formatDuration($totalDuration));
}
out("═══════════════════════════════════════════════════════════");
out("");
out("Next steps:");
out("  1. Verify row counts above match expectations");
out("  2. Check FK integrity (should say PASSED)");
out("  3. Set correct file permissions:");
out("     chown www-data:www-data {$config['sqlite']['path']}");
out("     chmod 664 {$config['sqlite']['path']}");
out("  4. Verify schema version:");
out("     sqlite3 {$config['sqlite']['path']} \"SELECT val FROM opts WHERE opt = 'dbversion';\"");
out("  5. Test the Statistics admin page in DokuWiki");
out("");

MysqlSource.php

<?php

namespace DokuWikiMigration;

use mysqli;
use RuntimeException;

/**
 * MySQL data source for the statistics migration.
 *
 * Reads from the old stats_* tables and provides batch iterators.
 */
class MysqlSource
{
    private mysqli $db;
    private string $prefix;
    private int $batchSize;

    public function __construct(array $config, int $batchSize = 5000)
    {
        $this->prefix = $config['prefix'] ?? 'stats_';
        $this->batchSize = $batchSize;

        $this->db = new mysqli(
            $config['host'],
            $config['user'],
            $config['password'],
            $config['database'],
            $config['port'] ?? 3306
        );

        if ($this->db->connect_error) {
            throw new RuntimeException('MySQL connection failed: ' . $this->db->connect_error);
        }

        $this->db->set_charset($config['charset'] ?? 'utf8mb4');
        // Force a consistent collation to avoid "Illegal mix of collations"
        // when JOINing/UNIONing MyISAM tables with different collations
        $this->db->query("SET collation_connection = 'utf8_unicode_ci'");
    }

    public function close(): void
    {
        $this->db->close();
    }

    /**
     * Get total row count for a table.
     */
    public function count(string $table): int
    {
        if (!preg_match('/^[a-zA-Z_][a-zA-Z0-9_]*$/', $table)) {
            throw new RuntimeException("Invalid table name: $table");
        }
        $fullTable = $this->prefix . $table;
        $result = $this->db->query("SELECT COUNT(*) AS cnt FROM `$fullTable`");
        if (!$result) {
            throw new RuntimeException("COUNT failed on $fullTable: " . $this->db->error);
        }
        $row = $result->fetch_assoc();
        $result->free();
        return (int)$row['cnt'];
    }

    /**
     * Batch-iterate rows from a table.
     *
     * @param string $table  Table name without prefix
     * @param string $orderBy  Column to order by for stable pagination
     * @return \Generator<array>  Yields batches (arrays of associative rows)
     */
    public function batchRead(string $table, string $orderBy = 'id'): \Generator
    {
        // Validate identifiers to prevent SQL injection
        if (!preg_match('/^[a-zA-Z_][a-zA-Z0-9_]*$/', $table)) {
            throw new RuntimeException("Invalid table name: $table");
        }
        if (!preg_match('/^[a-zA-Z_][a-zA-Z0-9_]*$/', $orderBy)) {
            throw new RuntimeException("Invalid column name: $orderBy");
        }

        $fullTable = $this->prefix . $table;
        $offset = 0;

        while (true) {
            $sql = "SELECT * FROM `$fullTable` ORDER BY `$orderBy` LIMIT {$this->batchSize} OFFSET $offset";
            $result = $this->db->query($sql);
            if (!$result) {
                throw new RuntimeException("Query failed on $fullTable: " . $this->db->error);
            }

            $batch = [];
            while ($row = $result->fetch_assoc()) {
                $batch[] = $row;
            }
            $result->free();

            if (empty($batch)) {
                break;
            }

            yield $batch;
            $offset += $this->batchSize;

            if (count($batch) < $this->batchSize) {
                break;
            }
        }
    }

    /**
     * Execute an arbitrary SELECT and return all rows.
     */
    public function query(string $sql): array
    {
        $result = $this->db->query($sql);
        if (!$result) {
            throw new RuntimeException("Query failed: " . $this->db->error . " SQL: $sql");
        }
        $rows = [];
        while ($row = $result->fetch_assoc()) {
            $rows[] = $row;
        }
        $result->free();
        return $rows;
    }

    /**
     * Execute a SELECT and yield batches via server-side iteration.
     *
     * Use for very large result sets where the full query with LIMIT/OFFSET
     * would be too slow.
     */
    public function batchQuery(string $sql): \Generator
    {
        $result = $this->db->query($sql, MYSQLI_USE_RESULT);
        if (!$result) {
            throw new RuntimeException("Query failed: " . $this->db->error . " SQL: $sql");
        }

        $batch = [];
        while ($row = $result->fetch_assoc()) {
            $batch[] = $row;
            if (count($batch) >= $this->batchSize) {
                yield $batch;
                $batch = [];
            }
        }
        $result->free();

        if (!empty($batch)) {
            yield $batch;
        }
    }

    // ─── Convenience readers for each source table ───

    public function readIplocation(): \Generator
    {
        return $this->batchRead('iplocation', 'ip');
    }

    public function readHistory(): \Generator
    {
        return $this->batchRead('history', 'info');
    }

    public function readLogins(): \Generator
    {
        return $this->batchRead('logins', 'id');
    }

    public function readSessions(): \Generator
    {
        return $this->batchRead('session', 'session');
    }

    public function readAccess(): \Generator
    {
        return $this->batchRead('access', 'id');
    }

    public function readOutlinks(): \Generator
    {
        return $this->batchRead('outlinks', 'id');
    }

    public function readSearch(): \Generator
    {
        return $this->batchRead('search', 'id');
    }

    public function readSearchwords(): \Generator
    {
        return $this->batchRead('searchwords', 'sid');
    }

    public function readEdits(): \Generator
    {
        return $this->batchRead('edits', 'id');
    }

    public function readMedia(): \Generator
    {
        return $this->batchRead('media', 'id');
    }

    public function readRefseen(): \Generator
    {
        return $this->batchRead('refseen', 'ref_md5');
    }

    /**
     * Get all distinct users from access, edits, and logins tables.
     * Returns rows with: user, first_seen (earliest dt), domain (null).
     */
    public function readDistinctUsers(): array
    {
        $p = $this->prefix;
        $sql = "
            SELECT u.user, u.first_seen
            FROM (
                SELECT `user`, MIN(dt) AS first_seen FROM `{$p}access` WHERE `user` != '' GROUP BY `user`
                UNION
                SELECT `user`, MIN(dt) AS first_seen FROM `{$p}edits` WHERE `user` != '' GROUP BY `user`
                UNION
                SELECT `user`, MIN(dt) AS first_seen FROM `{$p}logins` WHERE `user` != '' GROUP BY `user`
            ) u
            GROUP BY u.user
        ";
        // We take the MIN of first_seen across all three unioned sources.
        // But the UNION already does per-table MIN. We need global MIN per user.
        // Use COLLATE to handle mixed collations across MyISAM tables
        $sql = "
            SELECT user, MIN(first_seen) AS first_seen
            FROM (
                SELECT `user` COLLATE utf8_unicode_ci AS user, MIN(dt) AS first_seen FROM `{$p}access` WHERE `user` != '' GROUP BY `user`
                UNION ALL
                SELECT `user` COLLATE utf8_unicode_ci AS user, MIN(dt) AS first_seen FROM `{$p}edits` WHERE `user` != '' GROUP BY `user`
                UNION ALL
                SELECT `user` COLLATE utf8_unicode_ci AS user, MIN(dt) AS first_seen FROM `{$p}logins` WHERE `user` != '' GROUP BY `user`
            ) combined
            GROUP BY user
        ";
        return $this->query($sql);
    }

    /**
     * Get distinct external/search referrers from stats_access with their first-seen dates.
     */
    public function readDistinctReferers(): array
    {
        $p = $this->prefix;
        $sql = "
            SELECT
                a.ref AS url,
                COALESCE(rs.dt, MIN(a.dt)) AS first_seen,
                a.ref_type
            FROM `{$p}access` a
            LEFT JOIN `{$p}refseen` rs ON rs.ref_md5 = a.ref_md5
            WHERE a.ref_type IN ('external', 'search')
              AND a.ref IS NOT NULL
              AND a.ref != ''
            GROUP BY a.ref
        ";
        return $this->query($sql);
    }

    /**
     * Build a map of session → browser info from the first access row per session.
     * Done as a separate step to avoid slow cross-collation JOINs.
     *
     * @return array<string, array> session_id → [ua, ua_info, ua_type, ua_ver, os, user]
     */
    public function buildBrowserInfoMap(): array
    {
        $p = $this->prefix;
        // Get the MIN(id) per session first, then fetch those rows
        $sql = "
            SELECT a1.session, a1.ua, a1.ua_info, a1.ua_type, a1.ua_ver, a1.os, a1.user
            FROM `{$p}access` a1
            INNER JOIN (
                SELECT session, MIN(id) AS min_id
                FROM `{$p}access`
                GROUP BY session
            ) a2 ON a1.id = a2.min_id
        ";
        $rows = $this->query($sql);
        $map = [];
        foreach ($rows as $row) {
            $map[$row['session']] = $row;
        }
        return $map;
    }

    /**
     * Read sessions from stats_session in batches (no JOIN).
     */
    public function readSessionsSimple(): \Generator
    {
        return $this->batchRead('session', 'session');
    }

    /**
     * Get all distinct session IDs referenced in edits/outlinks/media.
     * Orphan detection is done in PHP (SessionResolver) to avoid cross-collation JOINs.
     */
    public function readReferencedSessionIds(): array
    {
        $p = $this->prefix;
        $allSessions = [];
        foreach (['edits', 'outlinks', 'media'] as $t) {
            $rows = $this->query("SELECT DISTINCT `session` FROM `{$p}{$t}` WHERE `session` != ''");
            foreach ($rows as $row) {
                $allSessions[$row['session']] = true;
            }
        }
        return array_keys($allSessions);
    }

    /**
     * Get browser info for an orphan session from the first access row that references it.
     */
    public function getOrphanSessionInfo(string $sessionId): ?array
    {
        $p = $this->prefix;
        $stmt = $this->db->prepare(
            "SELECT ua, ua_info, ua_type, ua_ver, os, `user`, dt, uid
             FROM `{$p}access`
             WHERE session = ?
             ORDER BY id ASC
             LIMIT 1"
        );
        $stmt->bind_param('s', $sessionId);
        $stmt->execute();
        $result = $stmt->get_result();
        $row = $result->fetch_assoc();
        $stmt->close();
        return $row ?: null;
    }

    /**
     * Get the MySQL table prefix.
     */
    public function getPrefix(): string
    {
        return $this->prefix;
    }
}

RefererMapper.php

<?php

namespace DokuWikiMigration;

/**
 * Maps old referrer URLs (from stats_access.ref) to new referers.id in SQLite.
 *
 * Builds an in-memory lookup from URL → SQLite ID during Phase 3.
 * Also determines search engine from ref_type.
 */
class RefererMapper
{
    /** @var array<string, int> URL → referer ID */
    private array $urlToId = [];

    private SqliteTarget $target;

    public function __construct(SqliteTarget $target)
    {
        $this->target = $target;
    }

    /**
     * Pre-populate the map from distinct referrers extracted from MySQL.
     *
     * @param array $referers  Rows with: url, first_seen, ref_type
     */
    public function buildFromMysqlData(array $referers): void
    {
        foreach ($referers as $ref) {
            $url = $ref['url'];
            if (isset($this->urlToId[$url])) {
                continue;
            }

            $engine = null;
            if (($ref['ref_type'] ?? '') === 'search') {
                $engine = $this->detectSearchEngine($url);
            }

            $id = $this->target->insertReferer($url, $ref['first_seen'], $engine);
            $this->urlToId[$url] = $id;
        }
    }

    /**
     * Resolve a referrer URL to its SQLite ID.
     *
     * Returns null if the URL should not be mapped (internal refs, empty after filter).
     */
    public function resolve(string $url, string $refType): ?int
    {
        // Only external and search referrers get mapped
        if (!in_array($refType, ['external', 'search'], true)) {
            return null;
        }

        if (empty($url)) {
            return null;
        }

        if (isset($this->urlToId[$url])) {
            return $this->urlToId[$url];
        }

        // Late arrival: insert on demand
        $engine = ($refType === 'search') ? $this->detectSearchEngine($url) : null;
        $id = $this->target->insertReferer($url, date('Y-m-d H:i:s'), $engine);
        $this->urlToId[$url] = $id;
        return $id;
    }

    /**
     * Simple search engine detection from URL hostname.
     */
    private function detectSearchEngine(string $url): ?string
    {
        $host = parse_url($url, PHP_URL_HOST);
        if (!$host) return null;

        $host = strtolower($host);
        $engines = [
            'google'     => 'google',
            'bing'       => 'bing',
            'yahoo'      => 'yahoo',
            'duckduckgo' => 'duckduckgo',
            'yandex'     => 'yandex',
            'baidu'      => 'baidu',
            'ecosia'     => 'ecosia',
            'startpage'  => 'startpage',
            'qwant'      => 'qwant',
        ];

        foreach ($engines as $needle => $name) {
            if (strpos($host, $needle) !== false) {
                return $name;
            }
        }

        return 'unknown';
    }

    /**
     * Get the number of mapped referrers.
     */
    public function getMappedCount(): int
    {
        return count($this->urlToId);
    }
}

SessionResolver.php

<?php

namespace DokuWikiMigration;

/**
 * Resolves session references and creates stub sessions for orphans.
 *
 * Some tables (edits, outlinks, media) reference sessions that may not exist
 * in stats_session. This class ensures every referenced session has an entry
 * in the SQLite sessions table.
 */
class SessionResolver
{
    private SqliteTarget $target;
    private MysqlSource $source;

    /** @var array<string, bool> Known session IDs (already in SQLite) */
    private array $knownSessions = [];

    /** @var array<string, array> Browser info map (session → browser data) */
    private array $browserMap = [];

    /** Placeholder session for tables with no session info (e.g. old search) */
    public const PLACEHOLDER_SESSION = '__migration_placeholder__';

    private int $stubsCreated = 0;

    public function __construct(MysqlSource $source, SqliteTarget $target)
    {
        $this->source = $source;
        $this->target = $target;
    }

    /**
     * Register a session ID as known (already migrated).
     */
    public function markKnown(string $sessionId): void
    {
        $this->knownSessions[$sessionId] = true;
    }

    /**
     * Set the browser info map (built in Phase 2) for orphan enrichment.
     */
    public function setBrowserMap(array $map): void
    {
        $this->browserMap = $map;
    }

    /**
     * Ensure the placeholder session exists.
     */
    public function ensurePlaceholder(): void
    {
        if (isset($this->knownSessions[self::PLACEHOLDER_SESSION])) {
            return;
        }

        $this->target->insertSession([
            'session' => self::PLACEHOLDER_SESSION,
            'dt'      => '1970-01-01 00:00:00',
            'end'     => '1970-01-01 00:00:00',
            'uid'     => '__migration__',
            'user'    => null,
            'ua'      => 'Migration Placeholder',
            'ua_info' => 'Migration',
            'ua_type' => 'browser',
            'ua_ver'  => '0',
            'os'      => 'Unknown',
        ]);
        $this->knownSessions[self::PLACEHOLDER_SESSION] = true;
    }

    /**
     * Find and create stubs for all orphan sessions referenced in edits/outlinks/media.
     * Uses the browserMap (from Phase 2) instead of per-session MySQL queries.
     *
     * @return int Number of stub sessions created
     */
    public function resolveOrphans(): int
    {
        $referencedIds = $this->source->readReferencedSessionIds();
        $count = 0;

        foreach ($referencedIds as $sessionId) {
            if (isset($this->knownSessions[$sessionId])) {
                continue;
            }

            // Use browser info map if available (built from stats_access in Phase 2)
            $bi = $this->browserMap[$sessionId] ?? [];

            $this->target->insertSession([
                'session' => $sessionId,
                'dt'      => $bi['dt'] ?? '1970-01-01 00:00:00',
                'end'     => $bi['dt'] ?? '1970-01-01 00:00:00',
                'uid'     => '__orphan__',
                'user'    => !empty($bi['user']) ? $bi['user'] : null,
                'ua'      => $bi['ua'] ?? '',
                'ua_info' => $bi['ua_info'] ?? '',
                'ua_type' => $bi['ua_type'] ?? 'browser',
                'ua_ver'  => $bi['ua_ver'] ?? '',
                'os'      => $bi['os'] ?? '',
            ]);

            $this->knownSessions[$sessionId] = true;
            $count++;
        }

        $this->stubsCreated = $count;
        return $count;
    }

    /**
     * Ensure a session exists, creating a minimal stub if needed.
     * Used during pageview/edit/media migration when we encounter an unknown session.
     */
    public function ensure(string $sessionId): string
    {
        if (empty($sessionId)) {
            $this->ensurePlaceholder();
            return self::PLACEHOLDER_SESSION;
        }

        if (isset($this->knownSessions[$sessionId])) {
            return $sessionId;
        }

        // Use browser info map if available
        $bi = $this->browserMap[$sessionId] ?? [];

        $this->target->insertSession([
            'session' => $sessionId,
            'dt'      => $bi['dt'] ?? '1970-01-01 00:00:00',
            'end'     => $bi['dt'] ?? '1970-01-01 00:00:00',
            'uid'     => '__late_stub__',
            'user'    => !empty($bi['user']) ? $bi['user'] : null,
            'ua'      => $bi['ua'] ?? '',
            'ua_info' => $bi['ua_info'] ?? '',
            'ua_type' => $bi['ua_type'] ?? 'browser',
            'ua_ver'  => $bi['ua_ver'] ?? '',
            'os'      => $bi['os'] ?? '',
        ]);

        $this->knownSessions[$sessionId] = true;
        $this->stubsCreated++;
        return $sessionId;
    }

    public function getStubsCreated(): int
    {
        return $this->stubsCreated;
    }

    public function getKnownCount(): int
    {
        return count($this->knownSessions);
    }
}

SqliteTarget.php

<?php

namespace DokuWikiMigration;

use PDO;
use RuntimeException;

/**
 * SQLite target for the statistics migration.
 *
 * Handles inserts into the new normalized SQLite schema with batch transactions.
 */
class SqliteTarget
{
    private PDO $db;
    private int $batchSize;
    private bool $dryRun;
    private bool $inTransaction = false;

    /** @var array<string, \PDOStatement> Prepared statement cache */
    private array $stmtCache = [];

    public function __construct(string $path, int $batchSize = 5000, bool $dryRun = false)
    {
        if (!file_exists($path)) {
            throw new RuntimeException("SQLite database not found: $path");
        }

        $this->batchSize = $batchSize;
        $this->dryRun = $dryRun;

        $this->db = new PDO("sqlite:$path", null, null, [
            PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
            PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC,
        ]);

        // Disable FK checks during migration, enable WAL for performance
        $this->db->exec('PRAGMA foreign_keys = OFF');
        $this->db->exec('PRAGMA journal_mode = WAL');
        $this->db->exec('PRAGMA synchronous = NORMAL');
    }

    public function close(): void
    {
        if ($this->inTransaction) {
            $this->commit();
        }
        // Re-enable foreign keys
        $this->db->exec('PRAGMA foreign_keys = ON');
        $this->stmtCache = [];
    }

    public function getPdo(): PDO
    {
        return $this->db;
    }

    public function beginTransaction(): void
    {
        if (!$this->inTransaction && !$this->dryRun) {
            $this->db->beginTransaction();
            $this->inTransaction = true;
        }
    }

    public function commit(): void
    {
        if ($this->inTransaction && !$this->dryRun) {
            $this->db->commit();
            $this->inTransaction = false;
        }
    }

    /**
     * Clear a table before import (for idempotent reruns).
     */
    public function clearTable(string $table): void
    {
        if ($this->dryRun) return;
        $this->db->exec("DELETE FROM `$table`");
    }

    /**
     * Get row count of a target table.
     */
    public function count(string $table): int
    {
        $stmt = $this->db->query("SELECT COUNT(*) FROM `$table`");
        return (int)$stmt->fetchColumn();
    }

    /**
     * Run PRAGMA foreign_key_check and return violations.
     */
    public function checkForeignKeys(): array
    {
        $stmt = $this->db->query('PRAGMA foreign_key_check');
        return $stmt->fetchAll();
    }

    /**
     * Get a cached prepared statement.
     */
    private function prepare(string $sql): \PDOStatement
    {
        if (!isset($this->stmtCache[$sql])) {
            $this->stmtCache[$sql] = $this->db->prepare($sql);
        }
        return $this->stmtCache[$sql];
    }

    /**
     * Detect whether iplocation uses 'dt' or 'lastupd' as the date column.
     */
    private function getIplocationDateColumn(): string
    {
        if (!isset($this->stmtCache['__iplocation_col'])) {
            $stmt = $this->db->query("PRAGMA table_info(iplocation)");
            $cols = array_column($stmt->fetchAll(), 'name');
            $this->stmtCache['__iplocation_col'] = in_array('lastupd', $cols) ? 'lastupd' : 'dt';
        }
        return $this->stmtCache['__iplocation_col'];
    }

    // ─── Table-specific insert methods ───

    public function insertIplocation(array $row): void
    {
        if ($this->dryRun) return;
        // Column is named 'dt' in some versions, 'lastupd' in others
        $colName = $this->getIplocationDateColumn();
        $stmt = $this->prepare(
            "INSERT OR IGNORE INTO iplocation (ip, code, country, city, host, $colName)
             VALUES (:ip, :code, :country, :city, :host, :dt)"
        );
        $stmt->execute([
            ':ip'      => $row['ip'],
            ':code'    => $row['code'] ?? '',
            ':country' => $row['country'] ?? '',
            ':city'    => $row['city'] ?? '',
            ':host'    => $row['host'] ?? '',
            ':dt'      => $row['lastupd'] ?? $row['dt'] ?? date('Y-m-d H:i:s'),
        ]);
    }

    public function insertHistory(array $row): void
    {
        if ($this->dryRun) return;
        $stmt = $this->prepare(
            'INSERT OR IGNORE INTO history (info, dt, value)
             VALUES (:info, :dt, :value)'
        );
        // Old schema has DATE, new has DATETIME. Pad if needed.
        $dt = $row['dt'];
        if (strlen($dt) === 10) {
            $dt .= ' 00:00:00';
        }
        $stmt->execute([
            ':info'  => $row['info'],
            ':dt'    => $dt,
            ':value' => (int)$row['value'],
        ]);
    }

    public function insertLogin(array $row): void
    {
        if ($this->dryRun) return;
        $stmt = $this->prepare(
            'INSERT INTO logins (dt, ip, user, type)
             VALUES (:dt, :ip, :user, :type)'
        );
        $stmt->execute([
            ':dt'   => $row['dt'],
            ':ip'   => $row['ip'] ?? '',
            ':user' => $row['user'] ?? '',
            ':type' => $row['type'] ?? '',
        ]);
    }

    public function insertUser(string $user, string $firstSeen, ?string $domain = null): void
    {
        if ($this->dryRun) return;
        $stmt = $this->prepare(
            'INSERT OR IGNORE INTO users (user, dt, domain)
             VALUES (:user, :dt, :domain)'
        );
        $stmt->execute([
            ':user'   => $user,
            ':dt'     => $firstSeen,
            ':domain' => $domain,
        ]);
    }

    public function insertSession(array $row): void
    {
        if ($this->dryRun) return;
        $stmt = $this->prepare(
            'INSERT OR IGNORE INTO sessions (session, dt, end, uid, user, ua, ua_info, ua_type, ua_ver, os)
             VALUES (:session, :dt, :end, :uid, :user, :ua, :ua_info, :ua_type, :ua_ver, :os)'
        );
        $stmt->execute([
            ':session' => $row['session'],
            ':dt'      => $row['dt'],
            ':end'     => $row['end'] ?? $row['dt'],
            ':uid'     => $row['uid'] ?? '',
            ':user'    => $row['user'] ?? null,
            ':ua'      => $row['ua'] ?? '',
            ':ua_info' => $row['ua_info'] ?? '',
            ':ua_type' => $row['ua_type'] ?? 'browser',
            ':ua_ver'  => $row['ua_ver'] ?? '',
            ':os'      => $row['os'] ?? '',
        ]);
    }

    /**
     * Insert a referrer and return its ID.
     */
    public function insertReferer(string $url, string $dt, ?string $engine = null): int
    {
        if ($this->dryRun) return 0;
        $stmt = $this->prepare(
            'INSERT OR IGNORE INTO referers (url, dt, engine)
             VALUES (:url, :dt, :engine)'
        );
        $stmt->execute([
            ':url'    => $url,
            ':dt'     => $dt,
            ':engine' => $engine,
        ]);

        // Fetch the ID (whether just inserted or already existed)
        $idStmt = $this->prepare('SELECT id FROM referers WHERE url = :url');
        $idStmt->execute([':url' => $url]);
        return (int)$idStmt->fetchColumn();
    }

    public function insertPageview(array $row): void
    {
        if ($this->dryRun) return;
        $stmt = $this->prepare(
            'INSERT INTO pageviews (dt, ip, session, page, ref_id, screen_x, screen_y, view_x, view_y)
             VALUES (:dt, :ip, :session, :page, :ref_id, :screen_x, :screen_y, :view_x, :view_y)'
        );
        $stmt->execute([
            ':dt'       => $row['dt'],
            ':ip'       => $row['ip'] ?? '',
            ':session'  => $row['session'] ?? '',
            ':page'     => $row['page'] ?? '',
            ':ref_id'   => $row['ref_id'],
            ':screen_x' => (int)($row['screen_x'] ?? 0),
            ':screen_y' => (int)($row['screen_y'] ?? 0),
            ':view_x'   => (int)($row['view_x'] ?? 0),
            ':view_y'   => (int)($row['view_y'] ?? 0),
        ]);
    }

    public function insertOutlink(array $row): void
    {
        if ($this->dryRun) return;
        $stmt = $this->prepare(
            'INSERT INTO outlinks (dt, session, link, page)
             VALUES (:dt, :session, :link, :page)'
        );
        $stmt->execute([
            ':dt'      => $row['dt'],
            ':session' => $row['session'] ?? '',
            ':link'    => $row['link'] ?? '',
            ':page'    => $row['page'] ?? '',
        ]);
    }

    /**
     * Insert a search entry and return its new auto-increment ID.
     */
    public function insertSearch(array $row): int
    {
        if ($this->dryRun) return 0;
        $stmt = $this->prepare(
            'INSERT INTO search (dt, ip, session, query)
             VALUES (:dt, :ip, :session, :query)'
        );
        $stmt->execute([
            ':dt'      => $row['dt'],
            ':ip'      => $row['ip'] ?? '',
            ':session' => $row['session'] ?? '',
            ':query'   => $row['query'] ?? '',
        ]);
        return (int)$this->db->lastInsertId();
    }

    public function insertSearchword(int $sid, string $word): void
    {
        if ($this->dryRun) return;
        $stmt = $this->prepare(
            'INSERT OR IGNORE INTO searchwords (sid, word) VALUES (:sid, :word)'
        );
        $stmt->execute([':sid' => $sid, ':word' => $word]);
    }

    public function insertEdit(array $row): void
    {
        if ($this->dryRun) return;
        $stmt = $this->prepare(
            'INSERT INTO edits (dt, ip, session, page, type)
             VALUES (:dt, :ip, :session, :page, :type)'
        );
        $stmt->execute([
            ':dt'      => $row['dt'],
            ':ip'      => $row['ip'] ?? '',
            ':session' => $row['session'] ?? '',
            ':page'    => $row['page'] ?? '',
            ':type'    => $row['type'] ?? '',
        ]);
    }

    public function insertMedia(array $row): void
    {
        if ($this->dryRun) return;
        $stmt = $this->prepare(
            'INSERT INTO media (dt, ip, session, media, size, mime1, mime2, inline)
             VALUES (:dt, :ip, :session, :media, :size, :mime1, :mime2, :inline)'
        );
        $stmt->execute([
            ':dt'      => $row['dt'],
            ':ip'      => $row['ip'] ?? '',
            ':session' => $row['session'] ?? '',
            ':media'   => $row['media'] ?? '',
            ':size'    => (int)($row['size'] ?? 0),
            ':mime1'   => $row['mime1'] ?? '',
            ':mime2'   => $row['mime2'] ?? '',
            ':inline'  => (int)($row['inline'] ?? 0),
        ]);
    }
}