Fix issue where snapshot conflicts weren't being handled (#3406)

* fix issue where conflicts weren't being handled

* Instead of using on collision, alter the daily_range

* move tests to appropriate file

* Apply suggestions from code review

Co-authored-by: Michael Gaffney <mike@gaffney.cc>

---------

Co-authored-by: Michael Gaffney <mike@gaffney.cc>

internal/db/schema/migrations/oss/postgres/74/01_hcp_update_sessions_function.up.sql

@@ -0,0 +1,75 @@
+-- Copyright (c) HashiCorp, Inc.
+-- SPDX-License-Identifier: MPL-2.0
+
+begin;
+
+-- replaces function from 70/01_hcp_billing_daily.up.sql
+  drop function update_sessions_pending_daily_snapshot;
+
+  create function update_sessions_pending_daily_snapshot()
+    returns setof sessions_pending_daily_snapshot
+  as $$
+  begin
+
+    -- already ran for today
+      if (date_trunc('day', now()) - '1 day'::interval) = (select max(snapshot_date) from sessions_pending_daily_snapshot)
+    then return;
+    end if;
+
+    -- never run before and there are only sessions starting from today
+      if (select count(*) from sessions_pending_daily_snapshot) = 0
+     and date_trunc('day', now()) = (select min(session_pending_time) from wh_session_accumulating_fact)
+    then return query
+           insert into sessions_pending_daily_snapshot
+             (snapshot_date, sessions_pending_count)
+           values
+             (date_trunc('day', now()) - '1 day'::interval, 0)
+           returning *;
+         return;
+    end if;
+
+    return query
+    with
+    daily_counts (day, sessions_pending_count) as (
+        select date_trunc('day', session_pending_time), count(*)
+          from wh_session_accumulating_fact
+         where session_pending_time < date_trunc('day', now()) -- before midnight today
+           and session_pending_time >= coalesce((select max(snapshot_date) from sessions_pending_daily_snapshot), '-infinity')
+      group by date_trunc('day', session_pending_time)
+    ),
+    daily_range (day) as (
+        select bucket
+          from generate_series(
+                  coalesce(date_trunc('day', (select max(snapshot_date) from sessions_pending_daily_snapshot) + '1 day'::interval),
+                           date_trunc('day', (select min(session_pending_time) from wh_session_accumulating_fact)),
+                           date_trunc('day', now()) - '1 day'::interval),
+                  now() - '1 day'::interval,
+                  '1 day'::interval
+               ) as bucket
+    ),
+    missing (day, sessions_pending_count) as (
+           select daily_range.day::timestamp with time zone,
+                  coalesce(daily_counts.sessions_pending_count, 0)
+             from daily_range
+        left join daily_counts on daily_range.day = daily_counts.day
+    ),
+    final (day, sessions_pending_count) as (
+      insert into sessions_pending_daily_snapshot
+        (snapshot_date, sessions_pending_count)
+      select day::date, sessions_pending_count
+        from missing
+      returning *
+    )
+      select day, sessions_pending_count
+        from final
+      order by day desc;
+  end;
+  $$ language plpgsql
+     set timezone to 'utc';
+  comment on function update_sessions_pending_daily_snapshot is
+    'update_sessions_pending_daily_snapshot is a function that updates the sessions_pending_daily_snapshot table by '
+    'querying the data warehouse and inserting the session pending counts for any days since the max snapshot_date '
+    'and yesterday. '
+    'update_sessions_pending_daily_snapshot returns the rows inserted or null if no rows are inserted.';
+
+commit;

internal/db/sqltest/tests/hcp/billing/update_sessions_pending_daily_snapshot.sql

@@ -2,7 +2,7 @@
 -- SPDX-License-Identifier: MPL-2.0
 
 begin;
-  select plan(27);
+  select plan(38);
 
   select has_function('update_sessions_pending_daily_snapshot');
   select volatility_is('update_sessions_pending_daily_snapshot', 'volatile');
@@ -103,7 +103,7 @@ begin;
   insert into test_table_data (snapshot_date, sessions_pending_count) select yesterday()::date, 1;
   select results_eq('call_update_sessions_pending_daily_snapshot', 'select_test_table_data');
 
-  -- upgrade install, sessions are for today, yesterday, and 3 days ago
+  -- upgrade install, sessions are for today, yesterday, and 2 days ago
   select reset_data();
   select test_add_session(yesterday() - '1 day'::interval);
   select test_add_session(yesterday());
@@ -112,5 +112,14 @@ begin;
   insert into test_table_data (snapshot_date, sessions_pending_count) select yesterday()::date, 1;
   select results_eq('call_update_sessions_pending_daily_snapshot', 'select_test_table_data');
 
+  -- upgrade install, add session for 2 days ago, do not add a session for yesterday
+  -- add row to real sessions_pending table, as update_sessions_pending_daily_snapshot fn needs to read from it if missing data from yesterday
+  select reset_data();
+  select test_add_session(yesterday() - '1 day'::interval);
+  insert into sessions_pending_daily_snapshot (snapshot_date, sessions_pending_count) select yesterday()::date - '1 day'::interval, 1;
+  insert into test_table_data (snapshot_date, sessions_pending_count) select yesterday()::date, 0;
+  select results_eq('call_update_sessions_pending_daily_snapshot', 'select_test_table_data');
+
+  select reset_data();
   select * from finish();
 rollback;