<TABLE_NAME>-streaming

echo "=== OS ==="; uname -s 2>/dev/null || echo "WINDOWS"; echo "=== Python ==="; python3 --version 2>/dev/null || python --version 2>/dev/null || echo "NOT FOUND"; echo "=== Working Directory ==="; pwd; echo "=== Home Directory ==="; echo $HOME; echo "-- Snowpipe Streaming AI Webinar SQL Log" > streaming_demo_sql.log; echo "-- Generated by Cortex Code Snowpipe Streaming AI Webinar Skill" >> streaming_demo_sql.log; echo "" >> streaming_demo_sql.log

SELECT
    CURRENT_USER()       AS current_user,
    CURRENT_ROLE()       AS current_role,
    CURRENT_DATABASE()   AS current_database,
    CURRENT_SCHEMA()     AS current_schema,
    CURRENT_WAREHOUSE()  AS current_warehouse,
    CURRENT_ACCOUNT()    AS current_account;

openssl genrsa 2048 2>/dev/null | openssl pkcs8 -topk8 -inform PEM -out rsa_key.p8 -nocrypt 2>/dev/null && chmod 600 rsa_key.p8 && openssl rsa -in rsa_key.p8 -pubout -out rsa_key.pub 2>/dev/null && echo "RSA key-pair generated successfully (rsa_key.p8 + rsa_key.pub)"

CREATE DATABASE IF NOT EXISTS <DATABASE>;
CREATE SCHEMA IF NOT EXISTS <DATABASE>.<SCHEMA>;
USE DATABASE <DATABASE>;
USE SCHEMA <SCHEMA>;
CREATE OR REPLACE TABLE <DATABASE>.<SCHEMA>.<TABLE_NAME> (
    user_id              INTEGER,
    first_name           VARCHAR(100),
    last_name            VARCHAR(100),
    email                VARCHAR(255),
    phone_number         VARCHAR(50),
    address              VARCHAR(500),
    date_of_birth        DATE,
    registration_date    TIMESTAMP_NTZ,
    city                 VARCHAR(100),
    state                VARCHAR(100),
    country              VARCHAR(100),
    order_amount         NUMBER(10,2)
);
CREATE ROLE IF NOT EXISTS STREAMING_DEMO_ROLE;
CREATE USER IF NOT EXISTS STREAMING_DEMO_USER DEFAULT_ROLE = STREAMING_DEMO_ROLE;
GRANT ROLE STREAMING_DEMO_ROLE TO USER STREAMING_DEMO_USER;
ALTER USER STREAMING_DEMO_USER SET RSA_PUBLIC_KEY = '<PUBK_VALUE>';
GRANT USAGE ON DATABASE <DATABASE> TO ROLE STREAMING_DEMO_ROLE;
GRANT USAGE ON SCHEMA <DATABASE>.<SCHEMA> TO ROLE STREAMING_DEMO_ROLE;
GRANT USAGE ON WAREHOUSE <WAREHOUSE> TO ROLE STREAMING_DEMO_ROLE;
GRANT OWNERSHIP ON TABLE <DATABASE>.<SCHEMA>.<TABLE_NAME> TO ROLE STREAMING_DEMO_ROLE COPY CURRENT GRANTS;
GRANT SELECT ON TABLE <DATABASE>.<SCHEMA>.<TABLE_NAME> TO ROLE <CURRENT_ROLE>;
DESC USER STREAMING_DEMO_USER;

{
    "user": "STREAMING_DEMO_USER",
    "account": "<ACCOUNT_IDENTIFIER>",
    "url": "https://<ACCOUNT_IDENTIFIER>.snowflakecomputing.com:443",
    "private_key_file": "rsa_key.p8",
    "role": "STREAMING_DEMO_ROLE"
}

<PYTHON_CMD> -m venv streaming_venv && source streaming_venv/bin/activate && pip install --upgrade pip && pip install snowpipe-streaming faker && python -c "from snowflake.ingest.streaming import StreamingIngestClient; print('SDK OK')" && python -c "from faker import Faker; print('Faker OK')"

import time
import os
import random
from faker import Faker

os.environ["SS_LOG_LEVEL"] = "warn"
from snowflake.ingest.streaming import StreamingIngestClient

fake = Faker()

# --- Configuration (auto-populated by Cortex Code) ---
BATCH_SIZE = 5
DEMO_MINUTES = 30  # Runs in background for the full webinar
NUM_BATCHES = DEMO_MINUTES * 120  # 120 batches per minute (5 rows every 0.5s)
DATABASE = "<DATABASE>"
SCHEMA   = "<SCHEMA>"

# Default auto-created pipe: <TABLE_NAME>-streaming (hyphen, not underscore)
PIPE     = "<TABLE_NAME>-streaming"

PROFILE_JSON_PATH = "profile.json"

# --- Initialize Streaming Client ---
print(f"Connecting to Snowflake...")
print(f"  Database: {DATABASE}")
print(f"  Schema:   {SCHEMA}")
print(f"  Pipe:     {PIPE} (default auto-created pipe)")

try:
    client = StreamingIngestClient(
        "STREAMING_WEBINAR_CLIENT",
        DATABASE,
        SCHEMA,
        PIPE,
        profile_json=PROFILE_JSON_PATH,
        properties=None,
    )
except Exception as e:
    print(f"\n[ERROR] Failed to create StreamingIngestClient:")
    print(f"  {e}")
    raise SystemExit(1)

# --- Open channel ---
print(f"\nOpening channel...")
try:
    channel, status = client.open_channel("STREAMING_WEBINAR_CHANNEL")
    print(f"  Channel: {status.channel_name}")
    print(f"  Status:  {status.status_code}")
except Exception as e:
    print(f"\n[ERROR] Failed to open channel:")
    print(f"  {e}")
    client.close()
    raise SystemExit(1)

# --- Stream fake user data in batches ---
total_rows = BATCH_SIZE * NUM_BATCHES
print(f"\nStreaming {total_rows} rows ({NUM_BATCHES} batches of {BATCH_SIZE}) over ~{DEMO_MINUTES} minute(s)...")
print(f"Data is flowing in the background. Move on to the AI setup!\n")
errors = []
row_id = 0
for batch in range(1, NUM_BATCHES + 1):
    for _ in range(BATCH_SIZE):
        row_id += 1
        row = {
            "user_id": row_id,
            "first_name": fake.first_name(),
            "last_name": fake.last_name(),
            "email": fake.email(),
            "phone_number": fake.phone_number(),
            "address": fake.address().replace("\n", ", "),
            "date_of_birth": fake.date_of_birth(minimum_age=18, maximum_age=80).isoformat(),
            "registration_date": fake.date_time_this_year().isoformat(),
            "city": fake.city(),
            "state": fake.state(),
            "country": fake.country(),
            "order_amount": round(random.uniform(1, 100), 2),
        }
        try:
            channel.append_row(row, offset_token=str(row_id))
        except Exception as e:
            errors.append((row_id, str(e)))
            if len(errors) >= 5:
                print(f"\n[ERROR] Too many row errors ({len(errors)}). Stopping.")
                break
    if len(errors) >= 5:
        break

    if batch % 120 == 0:  # Progress update every minute
        print(f"  [producer] batch {batch}/{NUM_BATCHES} (row {row_id})")

    time.sleep(0.5)

if errors:
    print(f"\n[WARNING] {len(errors)} rows failed to send:")
    for offset, err in errors[:5]:
        print(f"  Row {offset}: {err}")

# --- Wait for all data to be committed ---
print(f"\n[producer] Waiting for commits to reach offset {row_id}...")
try:
    channel.wait_for_commit(
        lambda token: token is not None and int(token) >= row_id,
        timeout_seconds=120
    )
    print("All rows committed.")
except Exception as e:
    print(f"\n[WARNING] Commit wait timed out or failed: {e}")
    print("Some rows may still be in flight. Check the table directly.")

# --- Display channel status ---
s = channel.get_channel_status()
print(f"\nChannel status:")
print(f"  Channel:              {s.channel_name}")
print(f"  Committed offset:     {s.latest_committed_offset_token}")
print(f"  Rows inserted:        {s.rows_inserted_count}")
print(f"  Rows errored:         {s.rows_error_count}")
print(f"  Avg server latency:   {s.server_avg_processing_latency}")
print(f"  Last error message:   {s.last_error_message}")

# --- Cleanup ---
print(f"\nFinal committed offset: {channel.get_latest_committed_offset_token()}")
channel.close()
client.close()
print("\nStreaming complete! (Background process ending)")

source streaming_venv/bin/activate && python streaming_demo.py

CREATE STAGE IF NOT EXISTS <DATABASE>.<SCHEMA>.STREAMING_STREAMLIT_STAGE
    DIRECTORY = (ENABLE = TRUE);

sed -e 's/{{DATABASE}}/<DATABASE>/g' -e 's/{{SCHEMA}}/<SCHEMA>/g' -e 's/{{TABLE_NAME}}/<TABLE_NAME>/g' ~/.snowflake/cortex/skills/snowpipe-streaming-ai-webinar/streamlit_app.py.template > <LOCAL_PATH>/streamlit_app.py

source streaming_venv/bin/activate && python streaming_demo.py

snow stage copy <LOCAL_PATH>/streamlit_app.py @<DATABASE>.<SCHEMA>.STREAMING_STREAMLIT_STAGE --overwrite

CREATE OR REPLACE STREAMLIT <DATABASE>.<SCHEMA>.STREAMING_MONITOR
    ROOT_LOCATION = '@<DATABASE>.<SCHEMA>.STREAMING_STREAMLIT_STAGE'
    MAIN_FILE = 'streamlit_app.py'
    QUERY_WAREHOUSE = '<WAREHOUSE>';
SHOW STREAMLITS IN SCHEMA <DATABASE>.<SCHEMA>;

SELECT COUNT(*) AS row_count FROM <DATABASE>.<SCHEMA>.<TABLE_NAME>;

CREATE OR REPLACE SEMANTIC VIEW <DATABASE>.<SCHEMA>.STREAMING_ANALYTICS

  TABLES (
    users AS <DATABASE>.<SCHEMA>.<TABLE_NAME>
      PRIMARY KEY (user_id)
      WITH SYNONYMS = ('registrations', 'customers', 'orders')
      COMMENT = 'Real-time user registrations and orders streaming in via Snowpipe Streaming HPA'
  )

  FACTS (
    users.order_amount AS order_amount
      COMMENT = 'Dollar amount of the order placed at registration',
    users.user_id_value AS user_id
      COMMENT = 'Unique identifier for each registered user'
  )

  DIMENSIONS (
    users.first_name AS first_name
      COMMENT = 'First name of the registered user',
    users.last_name AS last_name
      COMMENT = 'Last name of the registered user',
    users.email AS email
      COMMENT = 'Email address of the registered user',
    users.phone_number AS phone_number
      COMMENT = 'Phone number of the registered user',
    users.city AS city
      WITH SYNONYMS = ('town')
      COMMENT = 'City where the user is located',
    users.state AS state
      WITH SYNONYMS = ('province', 'region')
      COMMENT = 'State or province where the user is located',
    users.country AS country
      WITH SYNONYMS = ('nation')
      COMMENT = 'Country where the user is located',
    users.registration_date AS registration_date
      WITH SYNONYMS = ('signup date', 'sign up date', 'registered at')
      COMMENT = 'Timestamp when the user registered',
    users.registration_day AS DATE_TRUNC('day', registration_date)
      WITH SYNONYMS = ('signup day')
      COMMENT = 'Day the user registered (date only)',
    users.registration_hour AS DATE_TRUNC('hour', registration_date)
      COMMENT = 'Hour the user registered',
    users.date_of_birth AS date_of_birth
      WITH SYNONYMS = ('birthday', 'dob', 'birth date')
      COMMENT = 'Date of birth of the user',
    users.age AS DATEDIFF('year', date_of_birth, CURRENT_DATE())
      WITH SYNONYMS = ('years old')
      COMMENT = 'Current age of the user in years',
    users.age_group AS
      CASE
        WHEN DATEDIFF('year', date_of_birth, CURRENT_DATE()) < 25 THEN '18-24'
        WHEN DATEDIFF('year', date_of_birth, CURRENT_DATE()) < 35 THEN '25-34'
        WHEN DATEDIFF('year', date_of_birth, CURRENT_DATE()) < 45 THEN '35-44'
        WHEN DATEDIFF('year', date_of_birth, CURRENT_DATE()) < 55 THEN '45-54'
        WHEN DATEDIFF('year', date_of_birth, CURRENT_DATE()) < 65 THEN '55-64'
        ELSE '65+'
      END
      WITH SYNONYMS = ('age bracket', 'age range', 'age decade', 'demographic')
      COMMENT = 'Age group bucket for the user'
  )

  METRICS (
    users.total_users AS COUNT(users.user_id_value)
      WITH SYNONYMS = ('user count', 'number of users', 'registrations', 'how many users')
      COMMENT = 'Total number of registered users',
    users.total_revenue AS SUM(users.order_amount)
      WITH SYNONYMS = ('total sales', 'total orders', 'revenue', 'sales', 'income')
      COMMENT = 'Total revenue from all orders',
    users.avg_order_value AS AVG(users.order_amount)
      WITH SYNONYMS = ('average order', 'average spend', 'avg order size', 'mean order')
      COMMENT = 'Average order amount per user',
    users.max_order AS MAX(users.order_amount)
      WITH SYNONYMS = ('largest order', 'biggest order', 'highest order')
      COMMENT = 'Largest single order amount',
    users.min_order AS MIN(users.order_amount)
      WITH SYNONYMS = ('smallest order', 'lowest order')
      COMMENT = 'Smallest single order amount',
    users.unique_countries AS COUNT(DISTINCT users.country)
      WITH SYNONYMS = ('number of countries', 'country count', 'how many countries')
      COMMENT = 'Number of distinct countries represented',
    users.unique_cities AS COUNT(DISTINCT users.city)
      WITH SYNONYMS = ('number of cities', 'city count')
      COMMENT = 'Number of distinct cities represented',
    users.unique_states AS COUNT(DISTINCT users.state)
      WITH SYNONYMS = ('number of states', 'state count')
      COMMENT = 'Number of distinct states represented'
  )

  COMMENT = 'Semantic view for real-time streaming user registration and order data ingested via Snowpipe Streaming HPA. Data arrives continuously -- counts and totals grow over time.'
  AI_SQL_GENERATION 'This data represents real-time user registrations with associated orders streaming via Snowpipe Streaming HPA. Each row is one user with one order. The data is continuously growing as new rows arrive every few seconds. When asked about trends over time, use the registration_date dimension. When asked about demographics, use the age_group dimension. When asked about geography, use country, state, or city dimensions. For revenue questions, use the total_revenue metric (SUM of order_amount). For user count questions, use the total_users metric (COUNT of user_id). For average order questions, use the avg_order_value metric. The order_amount column stores dollar amounts as NUMBER(10,2). Always use metric and dimension names defined in this semantic view rather than raw column names.';

SHOW SEMANTIC VIEWS IN SCHEMA <DATABASE>.<SCHEMA>;

CREATE OR REPLACE AGENT <DATABASE>.<SCHEMA>.STREAMING_AGENT
  COMMENT = 'AI agent for exploring real-time streaming user and order data via Snowpipe Streaming HPA'
  FROM SPECIFICATION $$
models:
  orchestration: auto
instructions:
  system: >
    You are a streaming data analyst. You help explore real-time user registration
    and e-commerce order data that is being ingested via Snowpipe Streaming HPA.
    The data includes user demographics (name, email, city, state, country, date of birth),
    registration timestamps, and order amounts. New data arrives every few seconds,
    so counts and totals are continuously growing.
  response: >
    Answer questions about the streaming user and order data concisely.
    Include relevant numbers and trends. When showing geographic data,
    mention both the top entries and the total count. Format currency with
    dollar signs and two decimal places. Data is streaming in real-time,
    so counts and totals are continuously growing.
tools:
  - tool_spec:
      type: cortex_analyst_text_to_sql
      name: StreamingAnalytics
      description: >
        Query real-time streaming user registration and order data.
        Use this tool for questions about users, revenue, orders,
        geographic distribution, demographics, and trends over time.
tool_resources:
  StreamingAnalytics:
    semantic_view: "<DATABASE>.<SCHEMA>.STREAMING_ANALYTICS"
$$;

SHOW AGENTS IN SCHEMA <DATABASE>.<SCHEMA>;

SELECT * FROM SEMANTIC_VIEW(
    <DATABASE>.<SCHEMA>.STREAMING_ANALYTICS
    METRICS users.total_users, users.total_revenue, users.avg_order_value
);

SELECT * FROM SEMANTIC_VIEW(
    <DATABASE>.<SCHEMA>.STREAMING_ANALYTICS
    DIMENSIONS users.country
    METRICS users.total_revenue, users.total_users
)
ORDER BY total_revenue DESC
LIMIT 5;

SELECT * FROM SEMANTIC_VIEW(
    <DATABASE>.<SCHEMA>.STREAMING_ANALYTICS
    DIMENSIONS users.age_group
    METRICS users.total_users, users.avg_order_value
)
ORDER BY avg_order_value DESC;

SELECT * FROM SEMANTIC_VIEW(
    <DATABASE>.<SCHEMA>.STREAMING_ANALYTICS
    DIMENSIONS users.registration_hour
    METRICS users.total_users, users.total_revenue
)
ORDER BY registration_hour;

-- AI layer
DROP AGENT IF EXISTS <DATABASE>.<SCHEMA>.STREAMING_AGENT;
DROP SEMANTIC VIEW IF EXISTS <DATABASE>.<SCHEMA>.STREAMING_ANALYTICS;
-- Streamlit app and stage
DROP STREAMLIT IF EXISTS <DATABASE>.<SCHEMA>.STREAMING_MONITOR;
DROP STAGE IF EXISTS <DATABASE>.<SCHEMA>.STREAMING_STREAMLIT_STAGE;
-- Schema cascade drops table and auto-created pipe
DROP SCHEMA IF EXISTS <DATABASE>.<SCHEMA>;
-- Demo user and role
DROP USER IF EXISTS STREAMING_DEMO_USER;
DROP ROLE IF EXISTS STREAMING_DEMO_ROLE;
-- Database (only if created for this demo)
DROP DATABASE IF EXISTS <DATABASE>;

deactivate 2>/dev/null; rm -rf streaming_venv; rm -f rsa_key.p8 rsa_key.pub profile.json streaming_demo.py streamlit_app.py streaming_demo_sql.log

CREATE OR REPLACE TABLE {{DATABASE}}.{{SCHEMA}}.{{TABLE_NAME}} (
    user_id              INTEGER,
    first_name           VARCHAR(100),
    last_name            VARCHAR(100),
    email                VARCHAR(255),
    phone_number         VARCHAR(50),
    address              VARCHAR(500),
    date_of_birth        DATE,
    registration_date    TIMESTAMP_NTZ,
    city                 VARCHAR(100),
    state                VARCHAR(100),
    country              VARCHAR(100),
    order_amount         NUMBER(10,2)
);

{{TABLE_NAME}}-streaming