CREATE TABLE mastodon (
    value TEXT
) WITH (
    connector = 'sse',
    format = 'raw_string',
    endpoint = 'http://mastodon.arroyo.dev/api/v1/streaming/public',
    events = 'update'
);

CREATE VIEW tags AS (
    SELECT btrim(unnest(tags), '"') as tag FROM (
        SELECT extract_json(value, '$.tags[*].name') AS tags
     FROM mastodon)
);

SELECT * FROM (
    SELECT *, ROW_NUMBER() OVER (
        PARTITION BY window
        ORDER BY count DESC) as row_num
    FROM (SELECT count(*) as count,
        tag,
        hop(interval '5 seconds', interval '15 minutes') as window
            FROM tags
        group by tag, window)) WHERE row_num <= 5;

CREATE TABLE page_views (
    userId INT,
    state TEXT
)  WITH (
    connector = 'kafka',
    type = 'source',
    bootstrap_servers = 'localhost:9092',
    topic = 'page_view_events',
    format = 'json'
);

CREATE TABLE suspicious (
    user_id INT,
) WITH (
    connector = 'kafka',
    type = 'sink',
    bootstrap_servers = 'localhost:9092',
    topic = 'suspicious',
    format = 'json'
);

INSERT INTO suspicious
    SELECT "userId" as suspicious_id
        FROM (
            SELECT "userId",
                tumble(interval '1 day') as window,
                COUNT(distinct state) as states
            FROM page_views
            GROUP BY 1, 2)
            WHERE states > 4;