reducers: Align datatypes with postgres, instead of Redis

server/polar/reducer_bucket/redis_store.py

@@ -2,6 +2,7 @@
 import uuid
 from collections.abc import Mapping, Sequence
 from datetime import datetime, timedelta
+from decimal import Decimal
 
 import structlog
 
@@ -33,7 +34,7 @@ def get_reducer_bucket_key(
 
 
 async def add_to_buckets(
-    redis: Redis, contributions: Mapping[BucketId, Sequence[float]]
+    redis: Redis, contributions: Mapping[BucketId, Sequence[Decimal]]
 ) -> None:
     """Adds the values to their buckets, skipping unique aggregations."""
     ttl = int(REDUCER_BUCKET_TTL.total_seconds())
@@ -47,14 +48,14 @@ async def add_to_buckets(
                 case AggregationFunction.cnt:
                     await pipe.incrby(key, len(values))
                 case AggregationFunction.sum:
-                    await pipe.incrbyfloat(key, sum(values))
+                    await pipe.incrbyfloat(key, float(sum(values)))
                 case AggregationFunction.avg:
                     await pipe.hincrby(key, "count", len(values))
-                    await pipe.hincrbyfloat(key, "sum", sum(values))
+                    await pipe.hincrbyfloat(key, "sum", float(sum(values)))
                 case AggregationFunction.min:
-                    await pipe.zadd(key, {"min": min(values)}, lt=True)
+                    await pipe.zadd(key, {"min": float(min(values))}, lt=True)
                 case AggregationFunction.max:
-                    await pipe.zadd(key, {"max": max(values)}, gt=True)
+                    await pipe.zadd(key, {"max": float(max(values))}, gt=True)
                 case AggregationFunction.unique:
                     log.info(
                         "Ignoring reducer bucket of unique aggregation",

server/polar/reducer_bucket/repository.py

@@ -155,9 +155,9 @@ async def add_contribution(
         external_customer_id: str | None,
         *,
         count: int,
-        total: float = 0,
-        minimum: float | None = None,
-        maximum: float | None = None,
+        total: Decimal = Decimal(0),
+        minimum: Decimal | None = None,
+        maximum: Decimal | None = None,
     ) -> None:
         """Adds to the customer identity's latest bucket, or to the next generation
         once it's sealed."""

server/polar/reducer_bucket/service.py

@@ -28,7 +28,7 @@
 
 REDUCER_BUCKET_SIZE = timedelta(minutes=5)
 
-type Contributions = dict[redis_store.BucketId, list[tuple[datetime, float]]]
+type Contributions = dict[redis_store.BucketId, list[tuple[datetime, Decimal]]]
 
 
 def get_reducer_bucket_start(timestamp: datetime) -> datetime:
@@ -127,7 +127,7 @@ async def add_to_closed_buckets(
                     continue
                 statistics = (
                     {
-                        "total": sum(values),
+                        "total": sum(values, Decimal(0)),
                         "minimum": min(values),
                         "maximum": max(values),
                     }
@@ -468,12 +468,12 @@ def _reduce_events_to_bucket_contributions(
             if not reducer.filter.matches(event):
                 continue
             aggregation = reducer.aggregation
-            value = 1.0
+            value = Decimal(1)
             if isinstance(aggregation, PropertyAggregation):
                 property_value = aggregation.get_value(event)
                 if property_value is None:
                     continue
-                value = property_value
+                value = Decimal(str(property_value))
             bucket_id = (
                 reducer,
                 get_reducer_bucket_start(event.timestamp),

server/tests/reducer_bucket/test_service.py

@@ -134,7 +134,7 @@ async def test_adds_to_closed_buckets(
         organization: Organization,
     ) -> None:
         previous_start = BUCKET_START - REDUCER_BUCKET_SIZE
-        for batch in ((3, 4), (2,)):
+        for batch in ((0.1, 0.2), (0.3,)):
             await reducer_bucket_service.rollup(
                 session,
                 redis,
@@ -150,8 +150,8 @@ async def test_adds_to_closed_buckets(
             )
 
         [bucket] = await get_buckets(session)
-        assert (bucket.generation, bucket.count, bucket.sum) == (1, 3, Decimal(9))
-        assert (bucket.min, bucket.max) == (Decimal(2), Decimal(4))
+        assert (bucket.generation, bucket.count, bucket.sum) == (1, 3, Decimal("0.6"))
+        assert (bucket.min, bucket.max) == (Decimal("0.1"), Decimal("0.3"))
 
     async def test_adds_float_values_to_open_bucket(
         self,
@@ -238,17 +238,17 @@ async def test_skips_organizations_without_syncable_reducers(
             (
                 PropertyAggregation(func=AggregationFunction.sum, property="tokens"),
                 lambda redis, key: redis.get(key),
-                b"17",
+                b"18",
             ),
             (
                 PropertyAggregation(func=AggregationFunction.avg, property="tokens"),
                 lambda redis, key: redis.hmget(key, ["count", "sum"]),
-                [b"6", b"17"],
+                [b"6", b"18"],
             ),
             (
                 PropertyAggregation(func=AggregationFunction.min, property="tokens"),
                 lambda redis, key: redis.zscore(key, "min"),
-                1.0,
+                1.5,
             ),
             (
                 PropertyAggregation(func=AggregationFunction.max, property="tokens"),
@@ -275,7 +275,7 @@ async def test_accumulates_across_calls(
             save_fixture, organization=organization, meters=[meter]
         )
 
-        for batch in ((1, 2, 3), (5, 4, 2)):
+        for batch in ((1.5, 2, 3), (5, 4.5, 2)):
             events = [
                 build_event(organization, metadata={"tokens": tokens})
                 for tokens in batch