feat: fix local migrations and kafka topics

Makefile

@@ -178,7 +178,7 @@ DEV_COMPOSE := $(COMPOSE) -f docker-compose.yml -f docker-compose.dev.yml
 # tables, and S3 bucket. localstack runs with PERSISTENCE=0, so those are
 # wiped on every restart and must be recreated before any service (incl.
 # the natively-run backend) touches KMS/Dynamo/S3.
-INFRA_SVCS  := postgres redis zookeeper kafka schema-registry \
+INFRA_SVCS  := postgres redis zookeeper kafka kafka-init schema-registry \
                mailpit localstack localstack-init stripe-mock cloud-tasks-emulator
 
 # Language services. The things you iterate on; recreated per worktree.

docker-compose.yml

@@ -48,6 +48,8 @@ x-worker-base: &worker-base
   depends_on:
     kafka:
       condition: service_healthy
+    kafka-init:
+      condition: service_completed_successfully
     redis:
       condition: service_healthy
     schema-registry:
@@ -132,6 +134,25 @@ services:
       retries: 12
       start_period: 30s
 
+  kafka-init:
+    image: confluentinc/cp-kafka:7.5.0
+    depends_on:
+      kafka:
+        condition: service_healthy
+    entrypoint: ["/bin/bash", "-lc"]
+    command: >
+      "for topic in
+        jobs:worker-events
+        tracking-events
+        email-events
+        warmup-events
+        w:10c8f5e4-1c39-5b2a-9c8b-3d2f0a8b1a01
+        w:10c8f5e4-1c39-5b2a-9c8b-3d2f0a8b1a02
+        w:10c8f5e4-1c39-5b2a-9c8b-3d2f0a8b1a03;
+      do
+        kafka-topics --bootstrap-server kafka:29092 --create --if-not-exists --topic \"$$topic\" --partitions 1 --replication-factor 1;
+      done"
+
   schema-registry:
     image: confluentinc/cp-schema-registry:7.5.0
     depends_on:
@@ -277,6 +298,8 @@ services:
         condition: service_healthy
       kafka:
         condition: service_healthy
+      kafka-init:
+        condition: service_completed_successfully
       schema-registry:
         condition: service_started
       cloud-tasks-emulator:
@@ -317,6 +340,8 @@ services:
         condition: service_healthy
       kafka:
         condition: service_healthy
+      kafka-init:
+        condition: service_completed_successfully
       schema-registry:
         condition: service_started
 
@@ -365,6 +390,8 @@ services:
     depends_on:
       kafka:
         condition: service_healthy
+      kafka-init:
+        condition: service_completed_successfully
       schema-registry:
         condition: service_started
 

internal/infrastructure/db/migrations/000056_plan_stripe_price_yearly.down.sql

@@ -0,0 +1,2 @@
+DROP INDEX IF EXISTS idx_plans_stripe_price_yearly;
+ALTER TABLE plans DROP COLUMN IF EXISTS stripe_price_id_yearly;

internal/infrastructure/db/migrations/000056_plan_stripe_price_yearly.up.sql

@@ -0,0 +1,7 @@
+-- Add a yearly Stripe price to plans so a single plan tier can be billed
+-- either monthly (stripe_price_id) or annually (stripe_price_id_yearly).
+-- The interval the customer picks in billing decides which price is used at
+-- checkout; the existing duration column still records the row's primary cadence.
+ALTER TABLE plans ADD COLUMN stripe_price_id_yearly VARCHAR(255);
+
+CREATE INDEX idx_plans_stripe_price_yearly ON plans(stripe_price_id_yearly);