outpost: send reducer bucket totals over the WebSocket

server/polar/customer_meter/repository.py

@@ -15,7 +15,7 @@
     SortingClause,
 )
 from polar.kit.repository.base import Options
-from polar.models import Customer, CustomerMeter, Meter
+from polar.models import Customer, CustomerMeter, Meter, MeterReducer
 
 from .sorting import CustomerMeterSortProperty
 
@@ -58,45 +58,18 @@ async def get_by_customer_and_meter(
         )
         return await self.get_one_or_none(statement)
 
-    async def get_by_organization_customer_and_meter(
-        self,
-        organization_id: UUID,
-        customer_id: UUID,
-        meter_id: UUID,
-        *,
-        options: Options = (),
-    ) -> CustomerMeter | None:
+    async def get_all_by_customer_and_reducer(
+        self, customer_id: UUID, reducer_id: UUID
+    ) -> Sequence[CustomerMeter]:
         statement = (
             self.get_base_statement()
-            .join(CustomerMeter.customer)
+            .join(MeterReducer, MeterReducer.meter_id == CustomerMeter.meter_id)
             .where(
-                Customer.organization_id == organization_id,
                 CustomerMeter.customer_id == customer_id,
-                CustomerMeter.meter_id == meter_id,
+                MeterReducer.reducer_id == reducer_id,
             )
-            .options(contains_eager(CustomerMeter.customer), *options)
         )
-        return await self.get_one_or_none(statement)
-
-    async def get_by_organization_external_customer_and_meter(
-        self,
-        organization_id: UUID,
-        external_customer_id: str,
-        meter_id: UUID,
-        *,
-        options: Options = (),
-    ) -> CustomerMeter | None:
-        statement = (
-            self.get_base_statement()
-            .join(CustomerMeter.customer)
-            .where(
-                Customer.organization_id == organization_id,
-                Customer.external_id == external_customer_id,
-                CustomerMeter.meter_id == meter_id,
-            )
-            .options(contains_eager(CustomerMeter.customer), *options)
-        )
-        return await self.get_one_or_none(statement)
+        return await self.get_all(statement)
 
     async def get_by_customer_and_meter_for_update(
         self,

server/polar/customer_meter/service.py

@@ -29,6 +29,7 @@
 from polar.models.event import EventSource
 from polar.outpost.stream import enqueue as enqueue_outpost_event
 from polar.postgres import AsyncSession
+from polar.reducer.repository import ReducerRepository
 from polar.worker import enqueue_job
 
 from .repository import CustomerMeterRepository
@@ -195,12 +196,20 @@ async def update_customer_meter(
         )
         customer_meter.last_balanced_event = last_event
 
-        enqueue_outpost_event(
-            customer.organization_id,
-            type="customer_meter",
-            customer_id=customer.id,
-            meter_id=meter.id,
+        # Temporary: reducer values carry the credits until credits are reducers.
+        reducer = await ReducerRepository.from_session(session).get_by_meter_id(
+            meter.id
         )
+        if reducer is not None:
+            enqueue_outpost_event(
+                customer.organization_id,
+                {
+                    "type": "reducer_bucket",
+                    "reducer_id": reducer.id,
+                    "customer_id": customer.id,
+                    "external_customer_id": customer.external_id,
+                },
+            )
 
         return await repository.update(customer_meter), True
 

server/polar/outpost/schemas.py

@@ -13,7 +13,7 @@ class MessageBase(BaseModel):
 
 class IncomingMessageType(StrEnum):
     configuration = "configuration"
-    customer_meter = "customer_meter"
+    meter_usage = "meter_usage"
 
 
 class IncomingMessageBase(MessageBase):
@@ -24,35 +24,33 @@ class ConfigurationIncomingMessage(IncomingMessageBase):
     type: typing.Literal[IncomingMessageType.configuration]
 
 
-class CustomerMeterIncomingMessagePayloadBase(MessageBase):
+class MeterUsageIncomingMessagePayloadBase(MessageBase):
     meter_id: UUID4
 
 
-class CustomerMeterIncomingMessageCustomerPayload(
-    CustomerMeterIncomingMessagePayloadBase
-):
+class MeterUsageIncomingMessageCustomerPayload(MeterUsageIncomingMessagePayloadBase):
     customer_id: UUID4
 
 
-class CustomerMeterIncomingMessageExternalCustomerPayload(
-    CustomerMeterIncomingMessagePayloadBase
+class MeterUsageIncomingMessageExternalCustomerPayload(
+    MeterUsageIncomingMessagePayloadBase
 ):
     external_customer_id: str
 
 
-type CustomerMeterIncomingMessagePayload = (
-    CustomerMeterIncomingMessageCustomerPayload
-    | CustomerMeterIncomingMessageExternalCustomerPayload
+type MeterUsageIncomingMessagePayload = (
+    MeterUsageIncomingMessageCustomerPayload
+    | MeterUsageIncomingMessageExternalCustomerPayload
 )
 
 
-class CustomerMeterIncomingMessage(IncomingMessageBase):
-    type: typing.Literal[IncomingMessageType.customer_meter]
-    payload: CustomerMeterIncomingMessagePayload
+class MeterUsageIncomingMessage(IncomingMessageBase):
+    type: typing.Literal[IncomingMessageType.meter_usage]
+    payload: MeterUsageIncomingMessagePayload
 
 
 type IncomingMessage = typing.Annotated[
-    ConfigurationIncomingMessage | CustomerMeterIncomingMessage, Discriminator("type")
+    ConfigurationIncomingMessage | MeterUsageIncomingMessage, Discriminator("type")
 ]
 
 IncomingMessageAdapter: TypeAdapter[IncomingMessage] = TypeAdapter(IncomingMessage)
@@ -63,7 +61,7 @@ class CustomerMeterIncomingMessage(IncomingMessageBase):
 
 class OutgoingMessageType(StrEnum):
     configuration = "configuration"
-    customer_meter = "customer_meter"
+    meter_usage = "meter_usage"
 
 
 class OutgoingMessageBase(MessageBase):
@@ -86,22 +84,26 @@ class ConfigurationOutgoingMessage(OutgoingMessageBase):
     payload: ConfigurationOutgoingMessagePayload
 
 
-class CustomerMeterOutgoingMessagePayload(MessageBase):
-    customer_id: UUID4
-    external_customer_id: str | None
+class MeterUsageOutgoingMessagePayload(MessageBase):
     meter_id: UUID4
-    consumed_units: float
+    customer_id: UUID4 | None
+    external_customer_id: str | None
+    count: int
+    sum: float
+    min: float | None
+    max: float | None
+    # Temporary: credits come from customer meters until credits are reducers.
     credited_units: int
-    balance: float
 
 
-class CustomerMeterOutgoingMessage(OutgoingMessageBase):
-    type: typing.Literal[OutgoingMessageType.customer_meter]
-    payload: CustomerMeterOutgoingMessagePayload
+class MeterUsageOutgoingMessage(OutgoingMessageBase):
+    type: typing.Literal[OutgoingMessageType.meter_usage]
+    payload: MeterUsageOutgoingMessagePayload
 
 
 type OutgoingMessage = typing.Annotated[
-    ConfigurationOutgoingMessage | CustomerMeterOutgoingMessage, Discriminator("type")
+    ConfigurationOutgoingMessage | MeterUsageOutgoingMessage,
+    Discriminator("type"),
 ]
 
 OutgoingMessageAdapter: TypeAdapter[OutgoingMessage] = TypeAdapter(OutgoingMessage)