insights/bin/start-rust-service

#!/usr/bin/env bash

set -euo pipefail

#
# Starts up "live" local builds of the insights/rust
# services for a faster local dev inner loop
#

# set exit strategy
trap "trap - SIGTERM && kill -- -$$" SIGINT SIGTERM EXIT

# ensure we have a service name
if [ -z "$1" ]; then
  echo "ERROR missing required argument: insights/rust service name" >&2
  echo "Usage: bin/start-rust-service <SERVICE_NAME>" >&2
  exit 1
fi
RS_SVC="$1"

# Default hosts — overridden by env vars (e.g. in sandbox containers)
DEFAULT_KAFKA=${KAFKA_HOSTS:-kafka:9092}
DEFAULT_REDIS=${REDIS_URL:-redis://redis7:6379}
DEFAULT_PG="postgres://insights:insights@${PGHOST:-db}:${PGPORT:-5432}/insights"

# set up env appropriate for each *known* service - add new ones here!
case "$RS_SVC" in
  property-defs-rs)
    SQLX_QUERY_LEVEL=${SQLX_QUERY_LEVEL:-warn}
    export RUST_LOG=debug,sqlx::query=$SQLX_QUERY_LEVEL,rdkafka=warn
    export RUST_BACKTRACE=1
    export KAFKA_HOSTS="$DEFAULT_KAFKA"
    export DATABASE_URL="${DATABASE_URL:-$DEFAULT_PG}"
    export PERSONFN_ADDR="${PERSONFN_ADDR:-http://127.0.0.1:50052}"
    export SKIP_WRITES=false
    export SKIP_READS=false
    export FILTER_MODE=opt-out
    export DEBUG=1
    ;;

  property-vals-rs)
    export RUST_LOG=debug,rdkafka=warn
    export RUST_BACKTRACE=1
    export KAFKA_HOSTS="$DEFAULT_KAFKA"
    export BIND_PORT=3303 # avoid local-dev clashes on 3302 (cymbal) and the property-defs-rs default
    export KAFKA_CONSUMER_OFFSET_RESET=earliest
    export DEBUG=1
    ;;

  capture)
    SQLX_QUERY_LEVEL=${SQLX_QUERY_LEVEL:-warn}
    export RUST_LOG=debug,sqlx::query=$SQLX_QUERY_LEVEL,rdkafka=warn
    export RUST_BACKTRACE=1
    export ADDRESS=0.0.0.0:3307 # avoid port conflict in local dev
    export KAFKA_HOSTS="$DEFAULT_KAFKA"
    export REDIS_URL="$DEFAULT_REDIS"
    export DATABASE_URL="${DATABASE_URL:-$DEFAULT_PG}"
    export CAPTURE_MODE=events
    export LOG_LEVEL=debug
    export DEBUG=1
    export KAFKA_ERROR_TRACKING_TOPIC=${KAFKA_ERROR_TRACKING_TOPIC:-ingestion-errortracking-main}
    # v1 capture sink (MSK, writes to same topics as v0)
    export CAPTURE_V1_SINKS=${CAPTURE_V1_SINKS:-msk}
    export CAPTURE_V1_SINK_MSK_KAFKA_HOSTS=${CAPTURE_V1_SINK_MSK_KAFKA_HOSTS:-$DEFAULT_KAFKA}
    export CAPTURE_V1_SINK_MSK_KAFKA_TOPIC_MAIN=${CAPTURE_V1_SINK_MSK_KAFKA_TOPIC_MAIN:-events_plugin_ingestion}
    export CAPTURE_V1_SINK_MSK_KAFKA_TOPIC_HISTORICAL=${CAPTURE_V1_SINK_MSK_KAFKA_TOPIC_HISTORICAL:-events_plugin_ingestion_historical}
    export CAPTURE_V1_SINK_MSK_KAFKA_TOPIC_OVERFLOW=${CAPTURE_V1_SINK_MSK_KAFKA_TOPIC_OVERFLOW:-events_plugin_ingestion_overflow}
    export CAPTURE_V1_SINK_MSK_KAFKA_TOPIC_DLQ=${CAPTURE_V1_SINK_MSK_KAFKA_TOPIC_DLQ:-events_plugin_ingestion_dlq}
    export CAPTURE_V1_SINK_MSK_KAFKA_TOPIC_EXCEPTION=${CAPTURE_V1_SINK_MSK_KAFKA_TOPIC_EXCEPTION:-ingestion-errortracking-main}
    export CAPTURE_V1_SINK_MSK_KAFKA_TOPIC_HEATMAP=${CAPTURE_V1_SINK_MSK_KAFKA_TOPIC_HEATMAP:-heatmaps_ingestion}
    export CAPTURE_V1_SINK_MSK_KAFKA_TOPIC_CLIENT_INGESTION_WARNING=${CAPTURE_V1_SINK_MSK_KAFKA_TOPIC_CLIENT_INGESTION_WARNING:-ingestion-clientwarnings-main-1}
    ;;

  capture-replay)
    SQLX_QUERY_LEVEL=${SQLX_QUERY_LEVEL:-warn}
    export RUST_LOG=debug,sqlx::query=$SQLX_QUERY_LEVEL,rdkafka=warn
    export RUST_BACKTRACE=1
    export ADDRESS=0.0.0.0:3306 # avoid port conflict in local dev
    export KAFKA_HOSTS="$DEFAULT_KAFKA"
    export KAFKA_TOPIC=session_recording_snapshot_item_events
    export KAFKA_OVERFLOW_TOPIC=session_recording_snapshot_item_overflow
    export REDIS_URL="$DEFAULT_REDIS"
    export DATABASE_URL="${DATABASE_URL:-$DEFAULT_PG}"
    export CAPTURE_MODE=recordings
    export LOG_LEVEL=debug
    export DEBUG=1
    export DEV_RUST_SVC_OVERRIDE=capture # we want to run the same "capture" binary in a separate mode
    ;;

  capture-ai)
    SQLX_QUERY_LEVEL=${SQLX_QUERY_LEVEL:-warn}
    export RUST_LOG=debug,sqlx::query=$SQLX_QUERY_LEVEL,rdkafka=warn
    export RUST_BACKTRACE=1
    export ADDRESS=0.0.0.0:3309 # avoid port conflict in local dev (3308 is used by llm-gateway)
    export KAFKA_HOSTS="$DEFAULT_KAFKA"
    export KAFKA_TOPIC=ai_events_ingestion
    export REDIS_URL="$DEFAULT_REDIS"
    export DATABASE_URL="${DATABASE_URL:-$DEFAULT_PG}"
    export CAPTURE_MODE=events
    export LOG_LEVEL=debug
    export DEBUG=1
    # AI endpoint S3 blob storage (MinIO)
    export AI_S3_BUCKET=ai-blobs
    export AI_S3_PREFIX=llma/
    export AI_S3_ENDPOINT=${AI_S3_ENDPOINT:-http://objectstorage:19000}
    export AI_S3_REGION=us-east-1
    export AI_S3_ACCESS_KEY_ID=object_storage_root_user
    export AI_S3_SECRET_ACCESS_KEY=object_storage_root_password
    export DEV_RUST_SVC_OVERRIDE=capture # we want to run the same "capture" binary in a separate mode
    ;;

  capture-logs)
    export RUST_LOG=debug,rdkafka=warn
    export RUST_BACKTRACE=1
    export BIND_HOST=0.0.0.0
    export BIND_PORT=4320 # Offset from 4318 to avoid conflict with otel-collector host port mapping
    export MANAGEMENT_BIND_PORT=3304 # Park alongside the other Rust services (3301-3309); 8080/8081 clash with local infra (temporal-ui)
    export KAFKA_HOSTS=localhost:9092
    export KAFKA_TOPIC=logs_ingestion
    export KAFKA_METRICS_TOPIC=metrics_ingestion
    ;;

  cymbal)
    # Only cymbal crates at debug; everything else at info to keep the dev
    # log readable. rdkafka stays explicitly at warn because its info-level
    # output is still noisy. Unconditional export — flox activate exports its
    # own RUST_LOG, so a conditional default would silently inherit that and
    # match no cymbal modules.
    export RUST_LOG=info,cymbal=debug,cymbal_proto=debug,rdkafka=warn
    export RUST_BACKTRACE=1
    # Default mode, but set explicitly now that cymbal is multi-mode.
    export CYMBAL_MODE=processing
    export OBJECT_STORAGE_BUCKET=insights
    export FRAME_CACHE_TTL_SECONDS=1
    export FRAME_RESULT_TTL_MINUTES=0
    export ASSIGNMENT_RULE_CACHE_TTL_SECS=0
    export GROUPING_RULE_CACHE_TTL_SECS=0
    export SUPPRESSION_RULE_CACHE_TTL_SECS=0
    export BIND_PORT=3302 # HACK! avoid local-dev clash with property-defs-rs port
    export ALLOW_INTERNAL_IPS="true"
    export PERSONS_URL="${PERSONS_URL:-postgres://insights:insights@${PGHOST:-db}:${PGPORT:-5432}/insights_persons}"
    export ISSUE_BUCKETS_REDIS_URL="$DEFAULT_REDIS"
    export SIGNALS_API_BASE_URL="${SIGNALS_API_BASE_URL:-http://localhost:8000}" # Used to emit signals, via the internal API
    export INTERNAL_API_SECRET="${INTERNAL_API_SECRET:-insights123}"
    # Processing always routes exception resolution through cymbal-resolution.
    export CYMBAL_REMOTE_RESOLUTION_HOST="${CYMBAL_REMOTE_RESOLUTION_HOST:-127.0.0.1}"
    export CYMBAL_REMOTE_RESOLUTION_PORT="${CYMBAL_REMOTE_RESOLUTION_PORT:-50061}"
    # Keep cadence tight on the dev box so routing reacts quickly to
    # whatever load you induce by hand.
    export CYMBAL_REMOTE_RESOLUTION_DNS_REFRESH_SECS="${CYMBAL_REMOTE_RESOLUTION_DNS_REFRESH_SECS:-5}"
    export CYMBAL_REMOTE_RESOLUTION_SUBSCRIBE_TICK_HINT_MS="${CYMBAL_REMOTE_RESOLUTION_SUBSCRIBE_TICK_HINT_MS:-500}"
    ;;

  cymbal-resolution)
    # The cymbal binary in resolution mode (gRPC symbol-resolution service that
    # cymbal offloads to). Resolution mode only needs Postgres and S3 — no
    # Kafka/Redis/signals — hence the narrower env than the cymbal block.
    export CYMBAL_MODE=resolution
    # Run the `cymbal` binary; mode is selected via CYMBAL_MODE above.
    export DEV_RUST_SVC_OVERRIDE=cymbal
    # Only cymbal crates at debug; everything else at info. Unconditional
    # export — flox activate exports its own RUST_LOG, so a conditional
    # default would silently inherit that and match no cymbal modules.
    export RUST_LOG=info,cymbal=debug,cymbal_proto=debug,rdkafka=warn
    export RUST_BACKTRACE=1
    # S3 / MinIO bucket for symbol-file fetches.
    export OBJECT_STORAGE_BUCKET=insights
    # Frame cache tuning — values are intentionally tiny in dev so cache
    # behaviour is observable without warming.
    export FRAME_CACHE_TTL_SECONDS=1
    export FRAME_RESULT_TTL_MINUTES=0
    # Source-map / debug-file fetches may target cluster-internal hosts.
    export ALLOW_INTERNAL_IPS="true"
    # Listen on loopback only in dev. 50061 is the contract default; 9106 keeps
    # us clear of the personinsights-* metrics ports (9100-9105) and flags-consumer.
    export GRPC_ADDRESS="${GRPC_ADDRESS:-127.0.0.1:50061}"
    export METRICS_PORT="${METRICS_PORT:-9106}"
    export INTERNAL_API_SECRET="${INTERNAL_API_SECRET:-insights123}"
    # Match the dev cymbal cadence so the load bus updates often.
    export SUBSCRIBE_TICK_INTERVAL_MS="${SUBSCRIBE_TICK_INTERVAL_MS:-500}"
    ;;

  cymbal-notifications)
    # The notifications mode starts Temporal lifecycle workflows for notifications from processing mode.
    export CYMBAL_MODE=notifications
    export DEV_RUST_SVC_OVERRIDE=cymbal
    export RUST_LOG=debug,rdkafka=warn
    export RUST_BACKTRACE=1
    export KAFKA_HOSTS="$DEFAULT_KAFKA"
    export KAFKA_CONSUMER_OFFSET_RESET="${KAFKA_CONSUMER_OFFSET_RESET:-earliest}"
    export METRICS_PORT="${METRICS_PORT:-9107}"
    export INSIGHTS_API_KEY="${CYMBAL_NOTIFICATIONS_INSIGHTS_API_KEY:-pk-localinsightsprojecttoken}"
    export INSIGHTS_ENDPOINT="${CYMBAL_NOTIFICATIONS_INSIGHTS_ENDPOINT:-http://localhost:3307}"
    export TEMPORAL_HOST="${TEMPORAL_HOST:-temporal}"
    ;;

  embedding-worker)
    export RUST_LOG=debug,rdkafka=warn
    export RUST_BACKTRACE=1
    export KAFKA_CONSUMER_GROUP=embedding-worker
    export KAFKA_CONSUMER_TOPIC="document_embeddings_input"
    export FRAME_CACHE_TTL_SECONDS=1
    export BIND_PORT=3305 # HACK! avoid local-dev clash with property-defs-rs port
    export ALLOW_INTERNAL_IPS="true"
    ;;

  cyclotron-janitor)
    SQLX_QUERY_LEVEL=${SQLX_QUERY_LEVEL:-warn}
    export RUST_LOG=debug,sqlx::query=$SQLX_QUERY_LEVEL
    export RUST_BACKTRACE=1
    export KAFKA_HOSTS="$DEFAULT_KAFKA"
    export KAFKA_TOPIC=datastore_app_metrics2
    export DATABASE_URL="${CYCLOTRON_DATABASE_URL:-postgres://insights:insights@${PGHOST:-db}:${PGPORT:-5432}/cyclotron}"
    export BIND_PORT=3310 # default 3303 collides with property-vals-rs in local dev
    export DEBUG=1
    export ALLOW_INTERNAL_IPS=${ALLOW_INTERNAL_IPS:-true}
    ;;

  feature-flags)
    # Download MaxMind DB if it doesn't exist
    ./bin/download-mmdb

    export RUST_LOG=debug,tower=warn,maxminddb=warn
    export RUST_BACKTRACE=1
    export WRITE_DATABASE_URL="${WRITE_DATABASE_URL:-$DEFAULT_PG}"
    export READ_DATABASE_URL="${READ_DATABASE_URL:-$DEFAULT_PG}"
    export MAXMIND_DB_PATH=../share/GeoLite2-City.mmdb
    export REDIS_URL="$DEFAULT_REDIS"
    export FLAGS_REDIS_URL=${FLAGS_REDIS_URL:-redis://redis7:6379/1}
    export ADDRESS=0.0.0.0:3001
    export DEBUG=1
    export ALLOW_INTERNAL_IPS=${ALLOW_INTERNAL_IPS:-true}
    export INTERNAL_REQUEST_TOKEN=${INTERNAL_REQUEST_TOKEN:-dev-internal-token}
    ;;

  hypercache-server)
    export RUST_LOG=debug
    export RUST_BACKTRACE=1
    export REDIS_URL="$DEFAULT_REDIS"
    export ADDRESS=0.0.0.0:3002
    export DEBUG=1
    ;;

  batch-import-worker)
    SQLX_QUERY_LEVEL=${SQLX_QUERY_LEVEL:-warn}
    export RUST_LOG=debug,sqlx::query=$SQLX_QUERY_LEVEL,rdkafka=warn
    export RUST_BACKTRACE=1
    export KAFKA_HOSTS="$DEFAULT_KAFKA"
    export DATABASE_URL="${DATABASE_URL:-$DEFAULT_PG}"
    export BIND_HOST=0.0.0.0
    export BIND_PORT=3301
    export ENCRYPTION_KEYS=00beef0000beef0000beef0000beef00
    export KAFKA_TOPIC_MAIN=events_plugin_ingestion
    export KAFKA_TOPIC_HISTORICAL=events_plugin_ingestion_historical
    export KAFKA_TOPIC_OVERFLOW=events_plugin_ingestion_overflow
    export DEBUG=1
    ;;

  personinsights-replica)
    SQLX_QUERY_LEVEL=${SQLX_QUERY_LEVEL:-trace}
    export RUST_LOG=debug,sqlx::query=$SQLX_QUERY_LEVEL
    export RUST_BACKTRACE=1
    export GRPC_ADDRESS=127.0.0.1:50051
    export PRIMARY_DATABASE_URL="${PRIMARY_DATABASE_URL:-postgres://insights:insights@${PGHOST:-db}:${PGPORT:-5432}/insights_persons}"
    export METRICS_PORT=9100
    ;;

  personinsights-router)
    export RUST_LOG=debug
    export RUST_BACKTRACE=1
    export GRPC_ADDRESS=127.0.0.1:50052
    export REPLICA_URL=http://127.0.0.1:50051
    export BACKEND_TIMEOUT_MS=5000
    export METRICS_PORT=9101
    ;;

  personinsights-identity)
    SQLX_QUERY_LEVEL=${SQLX_QUERY_LEVEL:-trace}
    export RUST_LOG=debug,sqlx::query=$SQLX_QUERY_LEVEL
    export RUST_BACKTRACE=1
    export GRPC_ADDRESS=127.0.0.1:50055
    export PRIMARY_DATABASE_URL="${PRIMARY_DATABASE_URL:-postgres://insights:insights@${PGHOST:-db}:${PGPORT:-5432}/insights_persons}"
    export ROUTER_URL=http://127.0.0.1:50054
    export METRICS_PORT=9108
    ;;

  flags-consumer)
    SQLX_QUERY_LEVEL=${SQLX_QUERY_LEVEL:-warn}
    export RUST_LOG=debug,sqlx::query=$SQLX_QUERY_LEVEL,rdkafka=warn
    export RUST_BACKTRACE=1
    export FLAGS_READ_STORE_DATABASE_URL="${FLAGS_READ_STORE_DATABASE_URL:-postgres://insights:insights@${PGHOST:-db}:${PGPORT:-5432}/flags_read_store}"
    export KAFKA_HOSTS="${KAFKA_HOSTS:-localhost:9092}"
    export KAFKA_CONSUMER_OFFSET_RESET="${KAFKA_CONSUMER_OFFSET_RESET:-earliest}"
    export METRICS_PORT=9105
    export DEBUG=1
    ;;

  cohort-event-shuffler)
    SQLX_QUERY_LEVEL=${SQLX_QUERY_LEVEL:-warn}
    export RUST_LOG=debug,sqlx::query=$SQLX_QUERY_LEVEL,rdkafka=warn
    export RUST_BACKTRACE=1
    export KAFKA_HOSTS="$DEFAULT_KAFKA"
    export DATABASE_URL="${DATABASE_URL:-$DEFAULT_PG}" # reads insights_cohort for the realtime team gate
    # Code default gates to team 2 (the parity baseline); dev uses the local team, so open the gate.
    export REALTIME_COHORT_TEAM_ALLOWLIST="${REALTIME_COHORT_TEAM_ALLOWLIST:-all}"
    # earliest replays existing local events; prod defaults to "latest".
    export KAFKA_CONSUMER_OFFSET_RESET="${KAFKA_CONSUMER_OFFSET_RESET:-earliest}"
    export BIND_PORT=3310
    export DEBUG=1
    ;;

  cohort-stream-processor)
    SQLX_QUERY_LEVEL=${SQLX_QUERY_LEVEL:-warn}
    export RUST_LOG=debug,sqlx::query=$SQLX_QUERY_LEVEL,rdkafka=warn
    export RUST_BACKTRACE=1
    export DATABASE_URL="${DATABASE_URL:-$DEFAULT_PG}" # reads insights_cohort for the realtime filter catalog
    # Code default gates to team 2 (the parity baseline); dev uses the local team, so open the gate.
    export REALTIME_COHORT_TEAM_ALLOWLIST="${REALTIME_COHORT_TEAM_ALLOWLIST:-all}"
    # Fast refresh in dev so window/threshold edits surface a new LeafStateKey quickly.
    export FILTER_CATALOG_REFRESH_SECS="${FILTER_CATALOG_REFRESH_SECS:-30}"
    export FILTER_CATALOG_REFRESH_JITTER_SECS="${FILTER_CATALOG_REFRESH_JITTER_SECS:-5}"
    export BIND_PORT=3311
    export DEBUG=1
    ;;

  personinsights-router-leader)
    export RUST_LOG=debug
    export RUST_BACKTRACE=1
    export GRPC_ADDRESS=127.0.0.1:50054
    export REPLICA_URL=http://127.0.0.1:50051
    export ROUTER_MODE=leader
    export ETCD_ENDPOINTS=http://localhost:2379
    export ETCD_PREFIX=/personinsights/
    export BACKEND_TIMEOUT_MS=5000
    export METRICS_PORT=9104
    # The router binary is the same as personinsights-router
    export DEV_RUST_SVC_OVERRIDE=personinsights-router
    ;;

  personinsights-leader)
    export RUST_LOG=debug
    export RUST_BACKTRACE=1
    export GRPC_ADDRESS=127.0.0.1:50053
    export CACHE_MEMORY_CAPACITY=100000
    export ETCD_ENDPOINTS=http://localhost:2379
    export ETCD_PREFIX=/personinsights/
    # Register as 127.0.0.1 so the router can resolve the pod address locally.
    # In production, pod names resolve via K8s DNS.
    export POD_NAME=127.0.0.1
    export KAFKA_HOSTS="${KAFKA_HOSTS:-localhost:9092}"
    export KAFKA_PERSON_STATE_TOPIC=personinsights_updates
    export FALLBACK_DATABASE_URL="postgres://insights:insights@${PGHOST:-localhost}:${PGPORT:-5432}/insights_persons"
    export METRICS_PORT=9102
    ;;

  personinsights-writer)
    export RUST_LOG=debug
    export RUST_BACKTRACE=1
    export DATABASE_URL="postgres://insights:insights@${PGHOST:-localhost}:${PGPORT:-5432}/insights_persons"
    export KAFKA_HOSTS="${KAFKA_HOSTS:-localhost:9092}"
    export KAFKA_TOPIC=personinsights_updates
    export KAFKA_CONSUMER_GROUP=personinsights-writer
    export FLUSH_INTERVAL_MS=5000
    export FLUSH_BUFFER_SIZE=1000
    export BUFFER_CAPACITY=50000
    export METRICS_PORT=9103
    ;;

  *)
    echo "ERROR unknown insights/rust service '$RS_SVC' please register in: $0" >&2
    exit 1
    ;;
esac

# in some cases, we need to run the same underlying binary
# with different configuration - if a given RS_SVC config
# above sets the override env var, apply it here
if [[ -n "${DEV_RUST_SVC_OVERRIDE:-}" ]]; then
  export RS_SVC="$DEV_RUST_SVC_OVERRIDE"
fi

# ensure Rust services are in scope for "cargo run --bin <SERVICE_NAME>"
cd rust

# ensure the service restarts if it crashes in dev
sleep 1
while true; do
  if ! ps -p ${RS_PID:-} &>/dev/null; then
    cargo run --bin "$RS_SVC" &
    export RS_PID=$!
    echo -e "\n\033[47m>>\033[0m Restarted '$RS_SVC' with PID $RS_PID\n"
  fi

  # restart subprocess if it fails without bin/start shutdown
  set +e
  wait $RS_PID
  set -e

  sleep 3
done

echo -e "\n\033[47m>>\033[0m Shut down '$RS_SVC'\n"
exit 0