Singularity Checker

Overview

Singularity Checker 🎫

Singularity Checker is a bulk validation web application designed for operations and non-technical staff who need to verify large batches (10,000+) of promotional or redeem codes:

• Deduplication & Bulk Processing: Cleans and filters pasted inputs or uploaded files (.txt/.csv) to prevent redundant validations.
• Two-Stage Validation Engine: Leverages light, concurrent HTTP requests first, falling back to real browser automation only when required.
• Logged-In Session Support: Uses Playwright storage state JSON uploads to seamlessly validate behind user authentication.
• Persistent Job System: Runs validations asynchronously in the background, saving live results and progress metrics into a local database.
• Uncertain Rerunning: Lets users easily recheck only items that were blocked, rate-limited, or returned errors without restarting the entire job.

---

Key Features

• Bulk Upload: Pastes plain-text code lists or parses column-based .csv files automatically.
• Adaptive Concurrency Control: Allows users to fine-tune HTTP and browser worker counts, delays, and retries in real-time.
• Playwright Fallback Engine: Launches headless Chromium instances to interact with pages (forms, inputs, button clicks) and inspect dynamic layout elements.
• Profile-Driven Rules: Custom site validation logic is configured externally in declarative YAML profiles.
• Live Progress & Stats: Updates the browser dashboard dynamically using server-polled SQLite job updates.
• Detailed Export: Downloads final jobs into standardized CSV reports showing final code status, source (HTTP/Browser), reason, and response URLs.

---

Tech Stack

• Backend Framework: FastAPI + Uvicorn (Python 3.10+)
• Asynchronous HTTP Client: HTTPX
• Browser Automation: Playwright (Python API)
• Database Storage: SQLite (via aiosqlite in WAL mode)
• Declarative Profiles: PyYAML
• Frontend Layer: Vanilla HTML5, CSS3, and JavaScript

---

Architecture

Singularity Checker operates as a decoupled background worker system coordinating tasks via a database queue (SQLite) and dynamic async queues.

1. Verification State Machine Pipeline

Rather than launching heavy browser instances for every code, the system processes codes through a dual-stage state-machine flow. This minimizes overhead, avoiding resource bottlenecks and target-site rate limiting.

stateDiagram-v2
    [*] --> Pending : Bulk Upload (Deduplicated)
    
    state "Stage 1: Async HTTP Check" as HTTP
    Pending --> HTTP : Lightweight Async Clients
    
    state "Stage 2: Playwright Browser Fallback" as Browser
    HTTP --> Browser : Catch Blocked/Uncertain/Error (≤10% of codes)
    HTTP --> Valid : Success Rule Match (≥90% of codes)
    HTTP --> Invalid : Failure Rule Match
    
    Browser --> Valid : Content Success Match
    Browser --> Invalid : Content Failure Match
    Browser --> Blocked : Security/CAPTCHA Block
    Browser --> Error : Timeout/Network Exception
    
    Valid --> [*]
    Invalid --> [*]
    Blocked --> [*]
    Error --> [*]

2. SQLite as a Persistent Job Ledger

All jobs and code-level statuses are tracked in a SQLite database with WAL (Write-Ahead Logging) mode activated. This enables safe, concurrent write access as multiple async background workers update code check outcomes simultaneously.

3. Asynchronous Worker Coordination

The backend handles concurrent execution limits dynamically using asyncio.Queue workers. HTTP workers run concurrently up to the selected threshold and push failures or challenges directly into a secondary browser worker queue.

Below is the core loop from app/services/worker.py coordinating the lifetime of the concurrent worker pools:

# Create concurrent worker pools based on job configuration
http_workers = [
    asyncio.create_task(
        self._http_worker(job, profile, http_queue, browser_queue),
        name=f"http-worker-{job_id}-{index}",
    )
    for index in range(max(1, int(job["http_concurrency"])))
]

browser_enabled = bool(profile["browser"].get("enabled", True)) and int(job["browser_concurrency"]) > 0
browser_workers: list[asyncio.Task[None]] = []
if browser_enabled:
    browser_workers = [
        asyncio.create_task(
            self._browser_worker(job, profile, browser_queue),
            name=f"browser-worker-{job_id}-{index}",
        )
        for index in range(max(1, int(job["browser_concurrency"])))
    ]

# Await completion of HTTP stage, then safely signal workers to shut down
await http_queue.join()
for _ in http_workers:
    await http_queue.put(None)
await asyncio.gather(*http_workers, return_exceptions=True)

# Await completion of Browser stage and stop browser processes
if browser_workers:
    await browser_queue.join()
    for _ in browser_workers:
        await browser_queue.put(None)
    await asyncio.gather(*browser_workers, return_exceptions=True)