How to Detect Account Takeover in Real Time

Account takeover (ATO) is the endpoint of most credential theft campaigns. Attackers buy credential dumps, run automated stuffing against login endpoints, and pivot the successful logins into fraud within minutes. A detection system that runs batch jobs overnight catches nothing useful. You need streaming correlation across login, transaction, and device signals with sub-second latency.

RisingWave, an open-source PostgreSQL-compatible streaming database, lets you write ATO detection logic in plain SQL. The database reads from Kafka, continuously maintains materialized views over the event streams, and serves risk scores through its PostgreSQL interface -- no Java, no Flink cluster, no separate OLAP database.

What signals indicate account takeover?

ATO attacks leave three categories of signal: authentication anomalies, behavioral drift, and geographic impossibilities.

Authentication anomalies are the most direct. A burst of failed logins followed immediately by a success is the signature of a credential stuffing attack. Logins from newly registered devices or IP ranges associated with datacenter ASNs (indicating automation or proxies) are secondary confirmation.

Behavioral drift compares the current session to the user's historical pattern. An account that normally logs in from a single city during business hours and suddenly initiates a wire transfer at 3 AM from a mobile device is showing drift across multiple dimensions simultaneously.

Geographic impossibilities are the clearest signal. Two logins separated by more than 900 km/h of implied travel speed cannot both be the legitimate account owner.

How to build ATO detection with RisingWave

The architecture uses three Kafka topics as sources: login_events, transactions, and device_fingerprints. RisingWave ingests all three and correlates them through materialized views.

Step 1: Create the sources

CREATE SOURCE login_events (
    event_id      VARCHAR,
    user_id       VARCHAR,
    ip_address    VARCHAR,
    latitude      DOUBLE,
    longitude     DOUBLE,
    device_id     VARCHAR,
    success       BOOLEAN,
    event_time    TIMESTAMPTZ
) WITH (
    connector = 'kafka',
    topic = 'login_events',
    properties.bootstrap.server = 'kafka:9092',
    scan.startup.mode = 'latest'
) FORMAT PLAIN ENCODE JSON;

CREATE SOURCE transactions (
    txn_id        VARCHAR,
    user_id       VARCHAR,
    amount        DECIMAL,
    merchant_id   VARCHAR,
    txn_time      TIMESTAMPTZ
) WITH (
    connector = 'kafka',
    topic = 'transactions',
    properties.bootstrap.server = 'kafka:9092',
    scan.startup.mode = 'latest'
) FORMAT PLAIN ENCODE JSON;

Step 2: Detect credential stuffing with a tumbling window

A credential stuffing burst produces many failures in a short window before a success. Count failures and flag accounts where the failure rate in the last 10 minutes exceeds 5.

CREATE MATERIALIZED VIEW login_failure_rate AS
SELECT
    user_id,
    window_start,
    window_end,
    COUNT(*) FILTER (WHERE NOT success) AS failures,
    COUNT(*) FILTER (WHERE success) AS successes,
    COUNT(*) FILTER (WHERE NOT success)::FLOAT
        / NULLIF(COUNT(*), 0) AS failure_rate
FROM TUMBLE(login_events, event_time, INTERVAL '10 minutes')
GROUP BY user_id, window_start, window_end;

Step 3: Detect impossible travel

Compare consecutive logins per user using a self-join within a sliding window. The Haversine approximation for small distances works within SQL arithmetic.

CREATE MATERIALIZED VIEW impossible_travel AS
WITH consecutive AS (
    SELECT
        user_id,
        event_time AS t2,
        latitude   AS lat2,
        longitude  AS lon2,
        LAG(event_time)  OVER (PARTITION BY user_id ORDER BY event_time) AS t1,
        LAG(latitude)    OVER (PARTITION BY user_id ORDER BY event_time) AS lat1,
        LAG(longitude)   OVER (PARTITION BY user_id ORDER BY event_time) AS lon1
    FROM login_events
    WHERE success = TRUE
)
SELECT
    user_id,
    t1,
    t2,
    SQRT(
        POW(111.32 * (lat2 - lat1), 2) +
        POW(111.32 * COS(RADIANS((lat1 + lat2) / 2)) * (lon2 - lon1), 2)
    ) / NULLIF(EXTRACT(EPOCH FROM (t2 - t1)) / 3600.0, 0) AS speed_kmh
FROM consecutive
WHERE t1 IS NOT NULL;

CREATE MATERIALIZED VIEW impossible_travel_alerts AS
SELECT *
FROM impossible_travel
WHERE speed_kmh > 900;

Step 4: Correlate ATO signals into a risk score

Join the failure rate view and impossible travel alerts to compute a composite risk score per user.

CREATE MATERIALIZED VIEW ato_risk_scores AS
SELECT
    COALESCE(f.user_id, t.user_id) AS user_id,
    CURRENT_TIMESTAMP AS evaluated_at,
    CASE
        WHEN t.user_id IS NOT NULL                THEN 90
        WHEN f.failure_rate > 0.8 AND f.failures > 10 THEN 75
        WHEN f.failure_rate > 0.5                THEN 50
        ELSE 10
    END AS risk_score,
    f.failures,
    f.failure_rate,
    t.speed_kmh
FROM login_failure_rate f
FULL OUTER JOIN impossible_travel_alerts t
    ON f.user_id = t.user_id;

Your application queries ato_risk_scores over the PostgreSQL wire protocol. RisingWave keeps the materialized view continuously updated as new events arrive from Kafka.

Step 5: Sink high-risk events to an alert topic

CREATE SINK ato_alerts_sink
FROM ato_risk_scores
WHERE risk_score >= 75
WITH (
    connector = 'kafka',
    topic = 'ato_alerts',
    properties.bootstrap.server = 'kafka:9092'
) FORMAT PLAIN ENCODE JSON;

Why not use Apache Flink for ATO detection?

Apache Flink can detect ATO, but it imposes a significant operational tax. Writing the credential stuffing detector above in Flink requires Java or Scala DataStream API code, not SQL. Flink stores hot state in RocksDB on local disks, which requires careful checkpoint configuration to survive node failures. When your fraud team wants to adjust the speed threshold from 900 km/h to 800 km/h, Flink requires a job redeployment with state drain.

RisingWave handles all three pain points differently. Detection logic is SQL, so your security team can write and modify rules without an engineer. State lives in S3-compatible object storage, so it survives node failures automatically. Rule changes are ALTER MATERIALIZED VIEW statements that take effect on the next checkpoint without redeployment. And RisingWave runs 22 of 27 Nexmark benchmark streaming queries faster than Flink.

What is the latency of RisingWave ATO detection?

Detection latency is the time from a login event arriving in Kafka to the risk score being available for query. RisingWave processes events incrementally as they arrive and updates materialized views continuously. For the impossible travel view above, latency is typically under one second end-to-end, bounded by Kafka consumer poll intervals and the checkpoint cycle. Tumbling window views emit results at the end of each window period (10 minutes in the example above), which is the design of tumbling semantics.

How does this compare to rule-based fraud engines?

Traditional fraud engines like Kount or Sift evaluate rules one event at a time. They do not natively correlate events across streams or maintain stateful aggregates. You must build a separate pipeline to compute features, store them in a feature store, and join them at scoring time.

RisingWave collapses this architecture. The materialized view IS the feature store. The SQL join across login and transaction streams IS the correlation engine. There is no separate pipeline to maintain.

RisingWave is an open-source PostgreSQL-compatible streaming database. Read the account takeover detection landing page for architecture diagrams and deployment options.