fix(clickhouse): pay in advance handling (#2724)

This PR fixes the handling of events for pay in advance charges with
both `Clickhouse` and the ingest service

app/services/billable_metrics/aggregation_factory.rb

@@ -11,7 +11,7 @@ def supports_clickhouse?
     def self.new_instance(charge:, current_usage: false, **attributes)
       event_store = Events::Stores::PostgresStore
 
-      if !charge.pay_in_advance? && supports_clickhouse? && charge.billable_metric.organization.clickhouse_events_store?
+      if supports_clickhouse? && charge.billable_metric.organization.clickhouse_events_store?
         event_store = Events::Stores::ClickhouseStore
       end
 

app/services/events/common_factory.rb

@@ -12,7 +12,7 @@ def self.new_instance(source:)
           organization_id: source['organization_id'],
           transaction_id: source['transaction_id'],
           external_subscription_id: source['external_subscription_id'],
-          timestamp: Time.zone.at(source['timestamp'].to_f),
+          timestamp: Time.zone.at(source['timestamp'].to_i),
           code: source['code'],
           properties: source['properties']
         )

app/services/events/create_service.rb

@@ -52,7 +52,7 @@ def produce_kafka_event(event)
           external_subscription_id: event.external_subscription_id,
           transaction_id: event.transaction_id,
           # NOTE: Removes trailing 'Z' to allow clickhouse parsing
-          timestamp: event.timestamp.iso8601[...-1],
+          timestamp: event.timestamp.to_i,
           code: event.code,
           # NOTE: Default value to 0.0 is required for clickhouse parsing
           precise_total_amount_cents: event.precise_total_amount_cents.present? ? event.precise_total_amount_cents.to_s : "0.0",

spec/services/events/common_factory_spec.rb

@@ -22,7 +22,7 @@
         expect(new_instance.organization_id).to eq(source['organization_id'])
         expect(new_instance.transaction_id).to eq(source['transaction_id'])
         expect(new_instance.external_subscription_id).to eq(source['external_subscription_id'])
-        expect(new_instance.timestamp).to eq(Time.zone.at(source['timestamp'].to_f))
+        expect(new_instance.timestamp).to eq(Time.zone.at(source['timestamp'].to_i))
         expect(new_instance.code).to eq(source['code'])
         expect(new_instance.properties).to eq(source['properties'])
       end