replace kafka topic producers with send invocations

kafka is barely used and would just be an extra unnecessary dependency

accounting-service-api/src/lib/services.ts

@@ -1,12 +1,10 @@
 import { RestateService } from 'deepkit-restate';
 import { UUID } from '@deepkit/type';
 
-import { Consumer } from '@ftgo/consumer-service-api';
 import { Money } from '@ftgo/common';
 
 export interface AccountingServiceHandlers {
-  // Only required for handlers that need to be invoked directly using a Restate client
-  createAccount(consumer: Consumer): Promise<void>;
+  createAccount(consumerId: UUID): Promise<void>;
   authorize(
     consumerId: UUID,
     orderId: UUID,

accounting-service/src/accounting.service.ts

@@ -1,10 +1,9 @@
-import { restate, RestateServiceContext } from 'deepkit-restate';
+import { restate } from 'deepkit-restate';
 import { UUID } from '@deepkit/type';
 
-import { Consumer, KafkaConsumerTopic } from '@ftgo/consumer-service-api';
+import { Consumer } from '@ftgo/consumer-service-api';
 import { Money } from '@ftgo/common';
 import {
-  Account,
   AccountingServiceApi,
   AccountingServiceHandlers,
 } from '@ftgo/accounting-service-api';
@@ -15,10 +14,9 @@ import { AccountRepository } from './account.repository';
 export class AccountingService implements AccountingServiceHandlers {
   constructor(private readonly account: AccountRepository) {}
 
-  // @ts-ignore
-  @(restate.kafka<KafkaConsumerTopic>().handler())
-  async createAccount(consumer: Consumer): Promise<void> {
-    await this.account.create(consumer.id);
+  @restate.handler()
+  async createAccount(consumerId: UUID): Promise<void> {
+    await this.account.create(consumerId);
   }
 
   @restate.handler()
@@ -27,4 +25,13 @@ export class AccountingService implements AccountingServiceHandlers {
     orderId: UUID,
     orderTotal: Money,
   ): Promise<unknown> {}
+
+  @restate.handler()
+  async reverseAuthorization(
+    consumerId: UUID,
+    orderId: UUID,
+    orderTotal: Money,
+  ): Promise<unknown> {
+    return Promise.resolve(undefined);
+  }
 }

api-gateway/src/restaurant/restaurant.controller.ts

@@ -10,8 +10,8 @@ import {
 @http.controller('restaurant')
 export class RestaurantController {
   constructor(
-    private readonly service: RestaurantServiceApi,
     private readonly client: RestateClient,
+    private readonly service: RestaurantServiceApi,
   ) {}
 
   @http.POST('create')

consumer-service-api/src/index.ts

@@ -1,4 +1,3 @@
 export * from './lib/dtos';
 export * from './lib/entities';
 export * from './lib/services';
-export * from './lib/topics';

consumer-service/src/consumer.service.ts

@@ -1,13 +1,12 @@
-import { restate } from 'deepkit-restate';
-import { RestateKafkaProducer } from 'deepkit-restate/kafka';
+import { restate, RestateServiceContext } from 'deepkit-restate';
 import { UUID } from '@deepkit/type';
 
 import { Money, PersonName } from '@ftgo/common';
+import { AccountingServiceApi } from '@ftgo/accounting-service-api';
 import {
   Consumer,
   ConsumerServiceApi,
   ConsumerServiceHandlers,
-  KafkaConsumerTopic,
 } from '@ftgo/consumer-service-api';
 
 import { ConsumerRepository } from './consumer.repository';
@@ -16,13 +15,14 @@ import { ConsumerRepository } from './consumer.repository';
 export class ConsumerService implements ConsumerServiceHandlers {
   constructor(
     private readonly consumer: ConsumerRepository,
-    private readonly kafka: RestateKafkaProducer,
+    private readonly accounting: AccountingServiceApi,
+    private readonly ctx: RestateServiceContext,
   ) {}
 
   @restate.handler()
   async create(name: PersonName): Promise<UUID> {
     const consumer = (await this.consumer.create(name)) as Consumer;
-    await this.kafka.produce<KafkaConsumerTopic>([consumer]);
+    await this.ctx.send(this.accounting.createAccount(consumer.id));
     return consumer.id;
   }
 

consumer-service/src/main.ts

@@ -1,10 +1,11 @@
 import { App } from '@deepkit/app';
 import { FrameworkModule } from '@deepkit/framework';
 import { RestateKafkaProducerModule } from 'deepkit-restate/kafka';
-import { RestateModule } from 'deepkit-restate';
+import { provideRestateServiceProxy, RestateModule } from 'deepkit-restate';
 
-import { provideDatabase } from '@ftgo/common';
+import { AccountingServiceApi } from '@ftgo/accounting-service-api';
 import { Consumer } from '@ftgo/consumer-service-api';
+import { provideDatabase } from '@ftgo/common';
 
 import { ConsumerServiceConfig } from './config';
 import { ConsumerService } from './consumer.service';
@@ -22,7 +23,11 @@ void new App({
     }),
   ],
   controllers: [ConsumerService],
-  providers: [provideDatabase([Consumer]), ConsumerRepository],
+  providers: [
+    provideDatabase([Consumer]),
+    ConsumerRepository,
+    provideRestateServiceProxy<AccountingServiceApi>(),
+  ],
 })
   .setup((module, config: ConsumerServiceConfig) => {
     module

docker-compose.yml

@@ -15,16 +15,6 @@ services:
     volumes:
       - postgres:/var/lib/postgresql/data
 
-  redpanda:
-    image: docker.redpanda.com/redpandadata/redpanda:v24.2.2
-    ports:
-      - '8081:8081'
-      - '8082:8082'
-      - '9092:9092'
-      - '19644:9644'
-    volumes:
-      - redpanda:/var/lib/redpanda/data
-
   restate:
     image: docker.io/restatedev/restate:1.0.2
     ports:

infra/restate/config.prod.toml

@@ -10,7 +10,3 @@ bootstrap-num-partitions = 1
 [worker.invoker.retry-policy]
 initial-interval = "100ms"
 max-attempts = 5
-
-[[ingress.kafka-clusters]]
-name = "default"
-brokers = ["PLAINTEXT://ftgo-redpanda.internal:9092"]

infra/restate/config.toml

@@ -3,7 +3,3 @@ base-dir = "/var/lib/restate"
 [worker.invoker.retry-policy]
 initial-interval = "50ms"
 max-attempts = 3
-
-[[ingress.kafka-clusters]]
-name = "default"
-brokers = ["PLAINTEXT://redpanda:9092"]

kitchen-service-api/src/lib/services.ts

@@ -1,6 +1,8 @@
 import { RestateService } from 'deepkit-restate';
 import { UUID } from '@deepkit/type';
 
+import { Restaurant, RestaurantMenu } from '@ftgo/restaurant-service-api';
+
 import { Ticket, TicketDetails } from './entities';
 
 export interface KitchenServiceHandlers {
@@ -14,6 +16,8 @@ export interface KitchenServiceHandlers {
   beginCancelTicket(restaurantId: UUID, orderId: UUID): Promise<Ticket>;
   undoBeginCancelTicket(restaurantId: UUID, orderId: UUID): Promise<Ticket>;
   confirmCancelTicket(restaurantId: UUID, orderId: UUID): Promise<Ticket>;
+  reviseMenu(menu: RestaurantMenu): Promise<void>;
+  createMenu(restaurant: Restaurant): Promise<void>;
 }
 
 export type KitchenServiceApi = RestateService<

kitchen-service/src/kitchen.service.ts

@@ -1,24 +1,59 @@
 import { restate, RestateServiceContext } from 'deepkit-restate';
+import { UUID } from '@deepkit/type';
 
+import { Restaurant, RestaurantMenu } from '@ftgo/restaurant-service-api';
 import {
-  KafkaRestaurantMenuRevisedTopic,
-  KafkaRestaurantCreatedTopic,
-  Restaurant,
-  RestaurantMenu,
-} from '@ftgo/restaurant-service-api';
-import {
-  Kitchen,
   KitchenServiceApi,
   KitchenServiceHandlers,
+  Ticket,
+  TicketDetails,
 } from '@ftgo/kitchen-service-api';
 
 @restate.service<KitchenServiceApi>()
 export class KitchenService implements KitchenServiceHandlers {
-  // @ts-ignore
-  @(restate.kafka<KafkaRestaurantCreatedTopic>().handler())
+  @restate.handler()
   async createMenu(restaurant: Restaurant): Promise<void> {}
 
-  // @ts-ignore
-  @(restate.kafka<KafkaRestaurantMenuRevisedTopic>().handler())
+  @restate.handler()
   async reviseMenu(menu: RestaurantMenu): Promise<void> {}
+
+  @restate.handler()
+  async beginCancelTicket(restaurantId: UUID, orderId: UUID): Promise<Ticket> {
+    return Promise.resolve(undefined);
+  }
+
+  @restate.handler()
+  async cancelTicket(orderId: UUID): Promise<Ticket> {
+    return Promise.resolve(undefined);
+  }
+
+  @restate.handler()
+  async confirmCancelTicket(
+    restaurantId: UUID,
+    orderId: UUID,
+  ): Promise<Ticket> {
+    return Promise.resolve(undefined);
+  }
+
+  @restate.handler()
+  async confirmCreateTicket(ticketId: UUID): Promise<Ticket> {
+    return Promise.resolve(undefined);
+  }
+
+  @restate.handler()
+  async createTicket(
+    restaurantId: UUID,
+    orderId: UUID,
+    details: TicketDetails,
+  ): Promise<Ticket> {
+    return Promise.resolve(undefined);
+  }
+
+  @restate.handler()
+  async undoBeginCancelTicket(
+    restaurantId: UUID,
+    orderId: UUID,
+  ): Promise<Ticket> {
+    return Promise.resolve(undefined);
+  }
 }

order-service-api/src/lib/entities.ts

@@ -1,3 +1,5 @@
+import { ItemNotFound } from '@deepkit/orm';
+import { Writable } from 'type-fest';
 import {
   Embedded,
   entity,
@@ -8,7 +10,6 @@ import {
   PrimaryKey,
   UUID,
 } from '@deepkit/type';
-import { Writable } from 'type-fest';
 
 import {
   Address,
@@ -150,7 +151,13 @@ export class OrderLineItems {
   changeToOrderTotal(orderRevision: OrderRevision) {}
 
   find(menuItemId: UUID): OrderLineItem {
-    return this.lineItems.find(lineItem => lineItem.menuItemId === menuItemId);
+    const item = this.lineItems.find(
+      lineItem => lineItem.menuItemId === menuItemId,
+    );
+    if (!item) {
+      throw new ItemNotFound(menuItemId);
+    }
+    return item;
   }
 
   updateLineItems(orderRevision: OrderRevision) {}

order-service-api/src/lib/services.ts

@@ -1,7 +1,9 @@
-import { RestateService } from 'deepkit-restate';
+import { restate, RestateService } from 'deepkit-restate';
 import { UUID } from '@deepkit/type';
 import { ItemNotFound } from '@deepkit/orm';
 
+import { Restaurant, RestaurantMenu } from '@ftgo/restaurant-service-api';
+
 import { Order } from './entities';
 import { CreateOrderRequest } from './dtos';
 import { OrderMinimumNotMetException } from './replies';
@@ -16,6 +18,8 @@ export interface OrderServiceHandlers {
   confirmCancel(id: UUID): Promise<Order>;
   reject(id: UUID): Promise<Order>;
   approve(id: UUID): Promise<Order>;
+  createMenu(restaurant: Restaurant): Promise<void>;
+  reviseMenu(menu: RestaurantMenu): Promise<void>;
 }
 
 export type OrderServiceApi = RestateService<