Last-Mile Delivery Dispatch Backend: Assigning Drivers Without Double Booking

-- dispatch/migrations/1_create_dispatch.up.sql
CREATE TABLE drivers (
    id TEXT PRIMARY KEY,
    zone TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'off' -- off, available, busy
);

CREATE TABLE deliveries (
    id BIGSERIAL PRIMARY KEY,
    zone TEXT NOT NULL,
    pickup_lat DOUBLE PRECISION NOT NULL,
    pickup_lng DOUBLE PRECISION NOT NULL,
    status TEXT NOT NULL DEFAULT 'pending', -- pending, offered, accepted, delivered
    driver_id TEXT REFERENCES drivers(id),
    offer_expires_at TIMESTAMPTZ,
    declined_by TEXT[] NOT NULL DEFAULT '{}'
);
CREATE INDEX deliveries_open ON deliveries (status, offer_expires_at);

// dispatch/dispatch.go
package dispatch

import (
	"context"
	"errors"
	"math"
	"time"

	"encore.dev/cron"
	"encore.dev/pubsub"
	"encore.dev/storage/cache"
	"encore.dev/storage/sqldb"
)

var db = sqldb.NewDatabase("dispatch", sqldb.DatabaseConfig{Migrations: "./migrations"})

// Driver positions change every few seconds and only matter while fresh,
// so they live in Redis with a short expiry instead of in Postgres.
var positions = cache.NewCluster("positions", cache.ClusterConfig{
	EvictionPolicy: cache.VolatileLRU,
})

type Position struct {
	Lat, Lng float64
	At       time.Time
}

var lastPosition = cache.NewStructKeyspace[string, Position](positions, cache.KeyspaceConfig{
	KeyPattern:    "driver-pos/:key",
	DefaultExpiry: cache.ExpireIn(2 * time.Minute),
})

// Ping (omitted) is POST /drivers/:driverID/ping and writes lastPosition.

type DeliveryCreated struct {
	DeliveryID int64
}

var Created = pubsub.NewTopic[*DeliveryCreated]("delivery-created", pubsub.TopicConfig{
	DeliveryGuarantee: pubsub.AtLeastOnce,
})

type CreateParams struct {
	Zone                 string
	PickupLat, PickupLng float64
}

type CreateResponse struct {
	ID int64
}

// Create stores a delivery and hands it to the dispatcher.
//
//encore:api private method=POST path=/deliveries
func Create(ctx context.Context, p *CreateParams) (*CreateResponse, error) {
	var id int64
	err := db.QueryRow(ctx, `
		INSERT INTO deliveries (zone, pickup_lat, pickup_lng) VALUES ($1, $2, $3) RETURNING id
	`, p.Zone, p.PickupLat, p.PickupLng).Scan(&id)
	if err != nil {
		return nil, err
	}
	if _, err := Created.Publish(ctx, &DeliveryCreated{DeliveryID: id}); err != nil {
		return nil, err // the row stays pending and the minute sweep offers it
	}
	return &CreateResponse{ID: id}, nil
}

var _ = pubsub.NewSubscription(Created, "offer-to-driver", pubsub.SubscriptionConfig[*DeliveryCreated]{
	Handler: func(ctx context.Context, e *DeliveryCreated) error {
		return offer(ctx, e.DeliveryID)
	},
})

// offer picks the nearest available driver with a fresh position and offers them
// the delivery for 60 seconds. Both updates happen in one transaction, so a driver
// is never offered two deliveries and a delivery never goes to two drivers.
func offer(ctx context.Context, deliveryID int64) error {
	tx, err := db.Begin(ctx)
	if err != nil {
		return err
	}
	defer tx.Rollback()

	var zone string
	var lat, lng float64
	var declined []string
	err = tx.QueryRow(ctx, `
		SELECT zone, pickup_lat, pickup_lng, declined_by FROM deliveries
		WHERE id = $1 AND status = 'pending'
		FOR UPDATE SKIP LOCKED
	`, deliveryID).Scan(&zone, &lat, &lng, &declined)
	if errors.Is(err, sqldb.ErrNoRows) {
		return nil // already offered, accepted or being handled by another worker
	} else if err != nil {
		return err
	}

	rows, err := tx.Query(ctx, `
		SELECT id FROM drivers
		WHERE zone = $1 AND status = 'available' AND NOT (id = ANY($2))
		FOR UPDATE SKIP LOCKED
	`, zone, declined)
	if err != nil {
		return err
	}
	best, bestKm := "", math.MaxFloat64
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			rows.Close()
			return err
		}
		pos, err := lastPosition.Get(ctx, id)
		if errors.Is(err, cache.Miss) {
			continue // no ping in the last two minutes: treat as offline
		} else if err != nil {
			rows.Close()
			return err
		}
		if km := haversineKm(lat, lng, pos.Lat, pos.Lng); km < bestKm {
			best, bestKm = id, km
		}
	}
	rows.Close()
	if best == "" {
		return nil // nobody free: the sweep below retries every minute
	}

	if _, err := tx.Exec(ctx, `UPDATE drivers SET status = 'busy' WHERE id = $1`, best); err != nil {
		return err
	}
	if _, err := tx.Exec(ctx, `
		UPDATE deliveries SET status = 'offered', driver_id = $1, offer_expires_at = now() + interval '60 seconds'
		WHERE id = $2
	`, best, deliveryID); err != nil {
		return err
	}
	// ... push notification to the driver app ...
	return tx.Commit()
}

// Redispatch expires unanswered offers, frees those drivers and calls offer
// for every delivery still pending.
var _ = cron.NewJob("redispatch", cron.JobConfig{
	Title:    "Expire offers and redispatch",
	Every:    1 * cron.Minute,
	Endpoint: Redispatch,
})

//encore:api private
func Redispatch(ctx context.Context) error {
	if _, err := db.Exec(ctx, `
		WITH expired AS (
			UPDATE deliveries
			SET status = 'pending', declined_by = array_append(declined_by, driver_id),
			    driver_id = NULL, offer_expires_at = NULL
			WHERE status = 'offered' AND offer_expires_at < now()
			RETURNING declined_by[array_length(declined_by, 1)] AS driver_id
		)
		UPDATE drivers SET status = 'available' WHERE id IN (SELECT driver_id FROM expired)
	`); err != nil {
		return err
	}
	rows, err := db.Query(ctx, `SELECT id FROM deliveries WHERE status = 'pending' ORDER BY id LIMIT 500`)
	if err != nil {
		return err
	}
	defer rows.Close()
	for rows.Next() {
		var id int64
		if err := rows.Scan(&id); err != nil {
			return err
		}
		if err := offer(ctx, id); err != nil {
			return err
		}
	}
	return nil
}

func haversineKm(lat1, lng1, lat2, lng2 float64) float64 {
	const r = 6371.0
	dLat, dLng := (lat2-lat1)*math.Pi/180, (lng2-lng1)*math.Pi/180
	a := math.Sin(dLat/2)*math.Sin(dLat/2) + math.Cos(lat1*math.Pi/180)*math.Cos(lat2*math.Pi/180)*math.Sin(dLng/2)*math.Sin(dLng/2)
	return 2 * r * math.Asin(math.Sqrt(a))
}