AssignCustomerMiler and AssignCRMMiler retried five times, two minutes apart, using time.Sleep inside a bare goroutine — about ten minutes of state held only in one pod's memory. Any restart dropped every retry in flight, and nothing recorded it: the booking just stayed unassigned forever with no failure event, because publishAssignmentFailed only runs at the end of a loop that no longer existed. Deploying during a quiet patch was enough to lose bookings this way, and it happened during today's rollout. Retries now run on the ASSIGNMENTS stream. Each entry point publishes one booking.assignment_requested message; a durable consumer performs a single attempt per delivery and NAKs with retryDelay when no miler is available, so JetStream owns both the waiting and the delivery count. A pod dying mid-wait costs nothing — the message is still on the server and another replica takes it. Behaviour is deliberately unchanged from the caller's side: same five attempts, same two-minute spacing, same publishAssignmentFailed handoff to the DispatchAgent. The failure event is fired explicitly on the last delivery, since JetStream stops redelivering at MaxDeliver and would otherwise let the booking fail silently again. runInline keeps the old loop as a fallback for when JetStream is down. Assignment is how a booking reaches a rider, so it must not become dependent on the event bus: an outage should cost durability, which is what we had before, not stop bookings being assigned at all. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
182 lines
5.6 KiB
Go
182 lines
5.6 KiB
Go
package main
|
|
|
|
import (
|
|
"errors"
|
|
"os"
|
|
"os/signal"
|
|
"strings"
|
|
"syscall"
|
|
"time"
|
|
|
|
"doormile/config"
|
|
"doormile/controllers"
|
|
"doormile/db"
|
|
"doormile/internal/assignment"
|
|
"doormile/internal/notify"
|
|
"doormile/internal/worker"
|
|
"doormile/middlewares"
|
|
"doormile/migrations"
|
|
"doormile/routes"
|
|
"doormile/utils"
|
|
|
|
"github.com/gofiber/fiber/v2"
|
|
"github.com/gofiber/fiber/v2/middleware/cors"
|
|
"github.com/gofiber/fiber/v2/middleware/limiter"
|
|
"github.com/gofiber/fiber/v2/middleware/recover"
|
|
"github.com/joho/godotenv"
|
|
)
|
|
|
|
// errorHandler converts anything a handler returns — including a panic already
|
|
// turned into an error by the recover middleware — into the same
|
|
// {success, message} envelope the utils helpers emit, so clients never receive
|
|
// Fiber's default plain-text error body.
|
|
func errorHandler(c *fiber.Ctx, err error) error {
|
|
code := fiber.StatusInternalServerError
|
|
msg := "internal server error"
|
|
|
|
var fe *fiber.Error
|
|
if errors.As(err, &fe) {
|
|
code = fe.Code
|
|
msg = fe.Message
|
|
}
|
|
|
|
// 5xx means we broke, not the caller — log it with the route for triage.
|
|
if code >= fiber.StatusInternalServerError {
|
|
utils.Error("request failed",
|
|
"method", c.Method(),
|
|
"path", c.Path(),
|
|
"status", code,
|
|
"error", err.Error(),
|
|
)
|
|
}
|
|
|
|
return c.Status(code).JSON(fiber.Map{"success": false, "message": msg})
|
|
}
|
|
|
|
func main() {
|
|
// 1. Load configuration env variables
|
|
_ = godotenv.Load()
|
|
cfg := config.Load()
|
|
|
|
utils.Info("Starting Doormile Backend...")
|
|
|
|
// 2. Connect to Postgres, Redis & NATS
|
|
db.Connect(cfg)
|
|
db.InitRedis(cfg)
|
|
db.InitNATS(cfg)
|
|
notify.InitFCM()
|
|
|
|
// 3. Run GORM migrations for logistics tables
|
|
if db.DB != nil {
|
|
err := migrations.Migrate(db.DB)
|
|
if err != nil {
|
|
utils.Error("⚠️ Migration failed, continuing server boot...", "error", err.Error())
|
|
}
|
|
}
|
|
|
|
// 4. Initialize Fiber App
|
|
fiberCfg := fiber.Config{
|
|
AppName: "Doormile Logistics Service API v1",
|
|
ErrorHandler: errorHandler,
|
|
}
|
|
|
|
// Rate limiting keys on c.IP(). Behind a TLS-terminating reverse proxy the
|
|
// socket peer is the proxy, so without this every caller shares a single
|
|
// limit bucket and the whole fleet gets throttled together. Only honour
|
|
// X-Forwarded-For from proxies we explicitly trust — otherwise any client
|
|
// could spoof the header to dodge the limit entirely.
|
|
if cfg.TrustedProxies != "" {
|
|
proxies := strings.Split(cfg.TrustedProxies, ",")
|
|
for i := range proxies {
|
|
proxies[i] = strings.TrimSpace(proxies[i])
|
|
}
|
|
fiberCfg.EnableTrustedProxyCheck = true
|
|
fiberCfg.TrustedProxies = proxies
|
|
fiberCfg.ProxyHeader = fiber.HeaderXForwardedFor
|
|
utils.Info("Trusting X-Forwarded-For from configured proxies", "proxies", proxies)
|
|
} else {
|
|
utils.Warn("TRUSTED_PROXIES is unset — rate limits key on the socket peer address. " +
|
|
"If this service runs behind a reverse proxy, set TRUSTED_PROXIES or all clients will share one limit bucket.")
|
|
}
|
|
|
|
app := fiber.New(fiberCfg)
|
|
|
|
// Panic recovery — must be the outermost middleware so it also catches
|
|
// panics raised inside the ones registered below. Without this a single
|
|
// nil-pointer dereference in any handler takes the whole process down.
|
|
app.Use(recover.New(recover.Config{EnableStackTrace: true}))
|
|
|
|
// CORS policy
|
|
app.Use(cors.New(cors.Config{
|
|
AllowHeaders: "Origin,Content-Type,Accept,Authorization",
|
|
AllowOrigins: "http://localhost:5173,http://localhost:5174,http://localhost:3000,http://localhost:3001,http://localhost:3002,http://localhost:8080,http://localhost:8081,https://doormile.com,https://www.doormile.com,https://admin.doormile.com,https://api.doormile.com,https://crm.doormile.com,https://console.doormile.com,https://app.doormile.com,https://hub.doormile.com",
|
|
AllowCredentials: true,
|
|
AllowMethods: "GET,POST,PUT,DELETE,PATCH,OPTIONS",
|
|
}))
|
|
|
|
// Structured Zap logger middleware
|
|
app.Use(middlewares.ZapLogger())
|
|
|
|
// Global per-IP rate limit. Deliberately generous — this is an abuse
|
|
// backstop, not a quota. Auth endpoints get a much tighter limit of their
|
|
// own in routes.go. Health probes and websocket upgrades are exempt so
|
|
// orchestrator checks and long-lived tracking sockets are never throttled.
|
|
app.Use(limiter.New(limiter.Config{
|
|
Max: 300,
|
|
Expiration: 1 * time.Minute,
|
|
Next: func(c *fiber.Ctx) bool {
|
|
p := c.Path()
|
|
return strings.HasSuffix(p, "/health") ||
|
|
strings.HasSuffix(p, "/ready") ||
|
|
strings.HasPrefix(p, "/ws/")
|
|
},
|
|
LimitReached: func(c *fiber.Ctx) error {
|
|
return c.Status(fiber.StatusTooManyRequests).
|
|
JSON(fiber.Map{"success": false, "message": "too many requests, please slow down"})
|
|
},
|
|
}))
|
|
|
|
// 5. Register routes
|
|
routes.RegisterRoutes(app, cfg)
|
|
|
|
// 6. Pre-warm Redis pricing cache from Postgres
|
|
controllers.WarmPricingCache()
|
|
|
|
// 7. Start NATS booking worker in background
|
|
go worker.StartBookingWorker()
|
|
|
|
// 8. Start the assignment worker. Every replica runs one; they share a
|
|
// durable consumer, so JetStream hands each booking to exactly one of them.
|
|
go assignment.StartAssignmentWorker()
|
|
|
|
// 7. Startup server in a background thread
|
|
go func() {
|
|
utils.Info("Server starting", "port", cfg.Port)
|
|
if err := app.Listen(":" + cfg.Port); err != nil {
|
|
utils.Logger.Fatalf("Server failed to bind: %v", err)
|
|
}
|
|
}()
|
|
|
|
// Graceful shutdown listener
|
|
gracefulShutdown(app)
|
|
}
|
|
|
|
func gracefulShutdown(app *fiber.App) {
|
|
c := make(chan os.Signal, 1)
|
|
signal.Notify(c, os.Interrupt, syscall.SIGTERM)
|
|
|
|
<-c
|
|
utils.Info("Shutting down Doormile Backend...")
|
|
|
|
// Shutdown fiber
|
|
if err := app.Shutdown(); err != nil {
|
|
utils.Error("Fiber shutdown failed", "error", err)
|
|
}
|
|
|
|
// Close database pools
|
|
db.CloseDB()
|
|
|
|
time.Sleep(1 * time.Second)
|
|
utils.Info("Shutdown completed cleanly.")
|
|
}
|