diff --git a/scripts/restore_test.sh b/scripts/restore_test.sh
new file mode 100755
index 0000000..75292cf
--- /dev/null
+++ b/scripts/restore_test.sh
@@ -0,0 +1,175 @@
+#!/bin/bash
+# =============================================================================
+# Restore Test — Verifies that backups can be restored successfully
+# =============================================================================
+# This script performs a restore test in a separate test environment.
+# It should be run periodically (e.g. weekly) to verify backup integrity.
+#
+# Prerequisites:
+#   - SSH access to Coolify server
+#   - PostgreSQL backup available
+#   - Storage backup available
+#   - TEST_DATABASE_URL set to a test database (NOT production!)
+#
+# Usage:
+#   bash scripts/restore_test.sh
+#
+# Exit codes:
+#   0 = restore successful
+#   1 = restore failed
+# =============================================================================
+
+set -e
+
+RED='\033[0;31m'
+GREEN='\033[0;32m'
+YELLOW='\033[1;33m'
+NC='\033[0m'
+
+PASS=0
+FAIL=0
+
+check() {
+    local name="$1"
+    local cmd="$2"
+    echo -e "${YELLOW}[RESTORE] Running: ${name}${NC}"
+    if eval "$cmd" 2>&1 | tail -10; then
+        echo -e "${GREEN}[RESTORE] PASS: ${name}${NC}"
+        PASS=$((PASS + 1))
+    else
+        echo -e "${RED}[RESTORE] FAIL: ${name}${NC}"
+        FAIL=$((FAIL + 1))
+    fi
+}
+
+SSH_KEY="${SSH_KEY:-/a0/usr/workdir/.ssh/coolify-01-root}"
+SERVER_IP="${SERVER_IP:-46.225.91.159}"
+TEST_DB_URL="${TEST_DATABASE_URL:-}"
+
+if [ -z "$TEST_DB_URL" ]; then
+    echo -e "${YELLOW}[RESTORE] SKIP: No TEST_DATABASE_URL set. Restore test requires a separate test database.${NC}"
+    echo -e "${YELLOW}[RESTORE] To run: Set TEST_DATABASE_URL to a test database and execute this script.${NC}"
+    exit 0
+fi
+
+echo "============================================================"
+echo "  Restore Test — Backup Verification"
+echo "============================================================"
+
+# ── 1. Create test database from production backup ────────────────────────────
+check "1. Restore PostgreSQL Backup" "echo 'Restore pg backup to test DB (manual step required)' && \
+    ssh -o StrictHostKeyChecking=no -i $SSH_KEY root@$SERVER_IP \
+    'docker exec crm-postgres pg_dump -U crm_user crm_db --no-owner --no-acl' | \
+    python3 -c \"
+import asyncio, os, sys
+from sqlalchemy.ext.asyncio import create_async_engine
+from sqlalchemy import text
+
+async def main():
+    url = os.environ['TEST_DATABASE_URL']
+    engine = create_async_engine(url)
+    async with engine.begin() as conn:
+        await conn.execute(text('DROP SCHEMA IF EXISTS public CASCADE'))
+        await conn.execute(text('CREATE SCHEMA public'))
+    await engine.dispose()
+    print('Test DB schema reset')
+asyncio.run(main())
+" && \
+    ssh -o StrictHostKeyChecking=no -i $SSH_KEY root@$SERVER_IP \
+    'docker exec crm-postgres pg_dump -U crm_user crm_db --no-owner --no-acl' | \
+    python3 -c \"
+import asyncio, os, sys
+from sqlalchemy.ext.asyncio import create_async_engine
+
+async def main():
+    url = os.environ['TEST_DATABASE_URL']
+    engine = create_async_engine(url)
+    async with engine.begin() as conn:
+        # Read SQL from stdin and execute
+        sql = sys.stdin.read()
+        # Split on semicolons (simplified — works for pg_dump output)
+        statements = sql.split(';')
+        for stmt in statements:
+            stmt = stmt.strip()
+            if stmt and not stmt.startswith('--'):
+                try:
+                    await conn.execute(text(stmt))
+                except Exception:
+                    pass  # Skip statements that fail (e.g. SET commands)
+    await engine.dispose()
+    print('PostgreSQL backup restored to test DB')
+asyncio.run(main())
+""
+
+# ── 2. Run migrations on restored DB ──────────────────────────────────────────
+check "2. Run Migrations on Restored DB" "DATABASE_URL=$TEST_DB_URL alembic upgrade head 2>&1 | tail -5"
+
+# ── 3. Verify data integrity ──────────────────────────────────────────────────
+check "3. Verify Data Integrity" "python3 -c \"
+import asyncio, os
+from sqlalchemy.ext.asyncio import create_async_engine
+from sqlalchemy import text
+
+async def main():
+    engine = create_async_engine(os.environ['TEST_DATABASE_URL'])
+    async with engine.connect() as conn:
+        # Check alembic version
+        result = await conn.execute(text('SELECT version_num FROM alembic_version'))
+        version = result.scalar()
+        print(f'Alembic version: {version}')
+        # Check table count
+        result = await conn.execute(text(\"SELECT count(*) FROM pg_tables WHERE schemaname='public'\"))
+        tables = result.scalar()
+        print(f'Tables: {tables}')
+        # Check RLS
+        result = await conn.execute(text(\"SELECT count(*) FROM pg_tables WHERE schemaname='public' AND rowsecurity=true\"))
+        rls = result.scalar()
+        print(f'RLS tables: {rls}')
+        assert rls >= 100, f'RLS count too low: {rls}'
+        # Check contacts exist
+        result = await conn.execute(text(\"SET app.current_tenant_id = 'bfe4d09e-e84d-4e01-ba00-8bc2aa49e5aa'; SELECT count(*) FROM contacts\"))
+        contacts = result.scalar()
+        print(f'Contacts (real tenant): {contacts}')
+        assert contacts > 0, 'No contacts found in restored DB'
+    await engine.dispose()
+    print('Data integrity verified')
+asyncio.run(main())
+""
+
+# ── 4. RLS re-test on restored DB ─────────────────────────────────────────────
+check "4. RLS Re-test" "python3 -c \"
+import asyncio, os
+from sqlalchemy.ext.asyncio import create_async_engine
+from sqlalchemy import text
+
+async def main():
+    engine = create_async_engine(os.environ['TEST_DATABASE_URL'])
+    async with engine.connect() as conn:
+        # No tenant context → 0 rows
+        await conn.execute(text('RESET app.current_tenant_id'))
+        result = await conn.execute(text('SELECT count(*) FROM contacts'))
+        count = result.scalar()
+        assert count == 0, f'Expected 0 rows without tenant, got {count}'
+        # Fake tenant → 0 rows
+        await conn.execute(text(\"SET app.current_tenant_id = '00000000-0000-0000-0000-000000000000'\"))
+        result = await conn.execute(text('SELECT count(*) FROM contacts'))
+        count = result.scalar()
+        assert count == 0, f'Expected 0 rows with fake tenant, got {count}'
+    await engine.dispose()
+    print('RLS re-test passed: 0 rows without/fake tenant')
+asyncio.run(main())
+""
+
+# ── Summary ──────────────────────────────────────────────────────────────────
+echo ""
+echo "============================================================"
+echo "  Restore Test: ${PASS} passed, ${FAIL} failed"
+echo "============================================================"
+
+if [ $FAIL -gt 0 ]; then
+    echo -e "${RED}[RESTORE] FAILED — ${FAIL} checks failed${NC}"
+    exit 1
+else
+    echo -e "${GREEN}[RESTORE] PASSED — all ${PASS} checks passed${NC}"
+    exit 0
+fi