-- Create a Snowpipe for continuous ingestion
CREATE OR REPLACE PIPE orders_pipe
  AUTO_INGEST = TRUE
  AS
  COPY INTO raw_orders
  FROM @raw_stage/orders/
  FILE_FORMAT = (TYPE = 'PARQUET')
  MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE;

-- Bulk load with COPY INTO
COPY INTO raw_sales
FROM @my_s3_stage/sales/
FILE_FORMAT = (
  TYPE = 'CSV'
  FIELD_OPTIONALLY_ENCLOSED_BY = '"'
  SKIP_HEADER = 1
  NULL_IF = ('NULL', 'null', '')
)
ON_ERROR = 'SKIP_FILE';

-- Check pipe status
SELECT * FROM TABLE(INFORMATION_SCHEMA.COPY_ACTIVITY_HISTORY(
  PIPE_NAME => 'orders_pipe'
));

-- MERGE for incremental load with SCD Type 1
MERGE INTO dim_customers AS target
USING (
  SELECT
    customer_id,
    customer_name,
    email,
    phone,
    CURRENT_TIMESTAMP() AS load_date
  FROM raw_orders
  WHERE order_date >= '2024-01-01'
) AS source
ON target.customer_id = source.customer_id
WHEN MATCHED AND (
  target.customer_name != source.customer_name
  OR target.email != source.email
) THEN UPDATE SET
  customer_name = source.customer_name,
  email = source.email,
  phone = source.phone,
  load_date = source.load_date,
  is_current = TRUE
WHEN NOT MATCHED THEN INSERT (
  customer_id, customer_name, email, phone, load_date, is_current
) VALUES (
  source.customer_id,
  source.customer_name,
  source.email,
  source.phone,
  source.load_date,
  TRUE
);

-- Create a task for nightly ETL
CREATE OR REPLACE TASK nightly_etl_task
  WAREHOUSE = etl_wh
  SCHEDULE = 'USING CRON 0 2 * * * UTC'
  ALLOW_OVERLAPPING_EXECUTION = FALSE
AS
BEGIN
  -- Step 1: Load new orders
  COPY INTO raw_orders
  FROM @raw_stage/orders/
  FILE_FORMAT = (TYPE = 'PARQUET')
  ON_ERROR = 'SKIP_FILE';

  -- Step 2: Merge into dimension table
  MERGE INTO dim_orders AS target
  USING raw_orders AS source
  ON target.order_id = source.order_id
  WHEN MATCHED THEN UPDATE SET
    order_amount = source.order_amount,
    order_status = source.order_status
  WHEN NOT MATCHED THEN INSERT
    (order_id, customer_id, order_date, order_amount, order_status)
  VALUES
    (source.order_id, source.customer_id, source.order_date,
     source.order_amount, source.order_status);
END;

-- Create a task with a dependency chain
CREATE OR REPLACE TASK transform_customers
  WAREHOUSE = etl_wh
  AFTER = nightly_etl_task
AS
  MERGE INTO dim_customers AS target
  USING (SELECT DISTINCT customer_id, customer_name, email FROM raw_orders) AS source
  ON target.customer_id = source.customer_id
  WHEN NOT MATCHED THEN INSERT (customer_id, customer_name, email)
  VALUES (source.customer_id, source.customer_name, source.email);

-- Create a dynamic table for real-time aggregation
CREATE OR REPLACE DYNAMIC TABLE daily_sales_summary
  WAREHOUSE = analytics_wh
  TARGET_LAG = '1 hour'
AS
  SELECT
    DATE_TRUNC('day', order_date) AS sale_date,
    product_category,
    COUNT(DISTINCT order_id) AS order_count,
    SUM(order_amount) AS total_revenue,
    AVG(order_amount) AS avg_order_value
  FROM raw_orders o
  JOIN dim_products p ON o.product_id = p.product_id
  GROUP BY 1, 2;

-- Create a dynamic table for customer 360 view
CREATE OR REPLACE DYNAMIC TABLE customer_360
  WAREHOUSE = analytics_wh
  TARGET_LAG = '30 minutes'
AS
  SELECT
    c.customer_id,
    c.customer_name,
    c.email,
    COUNT(o.order_id) AS total_orders,
    SUM(o.order_amount) AS lifetime_value,
    MAX(o.order_date) AS last_order_date,
    DATEDIFF('day', MAX(o.order_date), CURRENT_DATE()) AS days_since_last_order
  FROM dim_customers c
  LEFT JOIN raw_orders o ON c.customer_id = o.customer_id
  GROUP BY 1, 2, 3;

-- Use Streams for CDC-based incremental loads
CREATE OR REPLACE STREAM raw_orders_stream
  ON TABLE raw_orders
  SHOW_INITIAL_ROWS = FALSE;

-- Task processes only new/changed rows
CREATE OR REPLACE TASK process_incremental_orders
  WAREHOUSE = etl_wh
  SCHEDULE = '5 MINUTE'
AS
  INSERT INTO dim_orders (order_id, customer_id, order_date, order_amount, status)
  SELECT order_id, customer_id, order_date, order_amount, status
  FROM raw_orders_stream
  WHERE METADATA$ACTION = 'INSERT';

-- Truncate and reload for small dimension tables
CREATE OR REPLACE TASK full_refresh_dim_products
  WAREHOUSE = etl_wh
  SCHEDULE = 'USING CRON 0 6 * * * UTC'
AS
BEGIN
  CREATE OR REPLACE TEMPORARY TABLE dim_products_new AS
  SELECT * FROM raw_products WHERE is_active = TRUE;

  TRUNCATE TABLE dim_products;
  INSERT INTO dim_products SELECT * FROM dim_products_new;
END;

-- SCD Type 2: Track historical changes
MERGE INTO dim_customers_scd2 AS target
USING raw_customers AS source
ON target.customer_id = source.customer_id
  AND target.is_current = TRUE
WHEN MATCHED AND (
  target.customer_name != source.customer_name
  OR target.email != source.email
) THEN UPDATE SET
  is_current = FALSE,
  end_date = CURRENT_DATE()
WHEN NOT MATCHED THEN INSERT
  (customer_id, customer_name, email, start_date, end_date, is_current)
VALUES
  (source.customer_id, source.customer_name, source.email,
   CURRENT_DATE(), NULL, TRUE);

-- Insert new version for changed records
INSERT INTO dim_customers_scd2
  (customer_id, customer_name, email, start_date, end_date, is_current)
SELECT
  customer_id, customer_name, email,
  CURRENT_DATE(), NULL, TRUE
FROM raw_customers r
WHERE EXISTS (
  SELECT 1 FROM dim_customers_scd2 d
  WHERE d.customer_id = r.customer_id
    AND d.is_current = FALSE
    AND d.end_date = CURRENT_DATE()
);