WITH daily_diff AS (
  SELECT
    product_id, snapshot_date, quantity,
    LAG(quantity, 1) OVER (               -- Retrieve the previous inventory quantity after sorting by date within each product
      PARTITION BY product_id
      ORDER BY     snapshot_date
    ) AS prev_quantity
  FROM stock_snapshots
)

SELECT
  product_id, snapshot_date, quantity, prev_quantity,
  quantity - prev_quantity AS diff           -- Calculate the difference between today and the previous day
FROM  daily_diff
WHERE prev_quantity IS NOT NULL              -- Exclude the first day (no previous-day data)
  AND quantity <> prev_quantity              -- Extract only rows where the inventory quantity changed
ORDER BY product_id, snapshot_date;

/*
  Logical execution order:
  [CTE: daily_diff]
  1. FROM stock_snapshots  → Read rows
  2. LAG()                 → Evaluate the window function (row count is preserved)

  [Main query]
  3. FROM daily_diff       → read the CTE
  4. WHERE                 → Filter rows
  5. SELECT                → Evaluate columns
  6. ORDER BY              → Sort and output
*/

WITH paged AS (
  SELECT
    order_id, customer_id, amount, status
  FROM  orders
  WHERE status   = 'pending'
    AND order_id > 1003       -- Cursor condition: use rows greater than the last ID retrieved previously
  ORDER BY order_id           -- Sort by the same column as the cursor
  LIMIT    5                  -- Maximum rows for the next page
)

SELECT
  *,
  MAX(order_id) OVER () AS next_cursor  -- Attach the last retrieved ID as the next-page cursor on every row
FROM paged;

/*
  Logical execution order:
  [CTE: paged]
  1. FROM orders   → Read rows
  2. WHERE         → Filter rows
  3. ORDER BY      → Sort and output
  4. LIMIT         → Limit the row count

  [Main query]
  5. FROM paged    → read the CTE
  6. SELECT        → Evaluate the window function (row count is preserved)
*/

-- ① CTE to retrieve the previous batch run time
WITH last_checkpoint AS (
  SELECT last_run_at
  FROM   batch_checkpoints
  WHERE  job_name = 'order_sync'
),
diff_summary AS (                         -- ② CTE to extract and aggregate changed records
  SELECT
    o.customer_id,
    COUNT(*)          AS order_count,     -- Order count
    SUM(o.amount)     AS total_amount,    -- Total amount
    MAX(o.updated_at) AS last_updated_at  -- Latest update time in the group
  FROM        orders o
  CROSS JOIN  last_checkpoint lc          -- Attach the previous run time (one row) to every row
  WHERE o.updated_at >= lc.last_run_at    -- Data updated since the previous batch run
    AND o.updated_at <  NOW()             -- Exclude new updates during this run and process them next time
  GROUP BY o.customer_id
)

-- Output the aggregate result
SELECT * FROM diff_summary ORDER BY customer_id;

-- ③ Update the checkpoint after the batch completes (run as a separate query)
UPDATE batch_checkpoints
SET    last_run_at = NOW()
WHERE  job_name    = 'order_sync';

/*
  Logical execution order for ②:
  1. FROM / CROSS JOIN  → Join (Cartesian product)
  2. WHERE              → Filter rows
  3. GROUP BY           → Group rows
  4. SELECT             → Evaluate aggregate functions
*/

WITH recent_runs AS (
  -- ① Number each job’s runs newest-first to select its latest 5 runs
  SELECT
    *,
    ROW_NUMBER() OVER (
      PARTITION BY job_name
      ORDER BY     started_at DESC
    ) AS rn
  FROM job_runs
)

SELECT
  job_name,
  COUNT(*) FILTER (WHERE status = 'success') AS success_count,          -- ② Count successes and failures separately
  COUNT(*) FILTER (WHERE status = 'failed')  AS fail_count,
  ROUND(                                                                -- ③ Calculate average duration in seconds and round to one decimal place
    AVG(EXTRACT(EPOCH FROM (ended_at - started_at)))
  , 1) AS avg_sec,
  BOOL_OR(                                                              -- ④ Stall detection: TRUE if even one run has been running for at least 30 minutes
    status = 'running'
    AND started_at < NOW() - INTERVAL '30 minutes'
  ) AS is_stalled,
  CASE                                                                  -- ⑤ Alert rule: alert when at least 3 of the latest 5 runs failed
    WHEN COUNT(*) FILTER (WHERE status = 'failed') >= 3 THEN '! ALERT'
    ELSE 'OK'
  END AS alert

FROM  recent_runs
WHERE rn <= 5
GROUP BY job_name
ORDER BY job_name;

/*
  Logical execution order:
  1. FROM / ROW_NUMBER  → Evaluate the window function (row count is preserved)
  2. FROM(CTE) / WHERE  → Filter rows
  3. GROUP BY           → Group rows
  4. SELECT             → Evaluate aggregate functions
  5. ORDER BY           → Sort and output
*/